mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
zeekay
d5f8e3f2b5
Introduces hanzo-tools-* namespace packages for independent tool installation: Core packages: - hanzo-tools-core: BaseTool, ToolRegistry, PermissionManager - hanzo-tools-filesystem: read, write, edit, tree, find, search, ast (7 tools) - hanzo-tools-shell: dag, ps, zsh, shell, npx, uvx, open (7 tools) AI/Agent packages: - hanzo-tools-agent: critic, iching, review, network, CLI agents (12 tools) - hanzo-tools-memory: recall/create/update/delete memories, facts, KB (9 tools) - hanzo-tools-reasoning: think, critic (2 tools) Editor/Dev packages: - hanzo-tools-lsp: Language Server Protocol tool (1 tool) - hanzo-tools-refactor: rename, extract, inline, move, change_signature (1 tool) - hanzo-tools-database: SQL + graph database tools (8 tools) - hanzo-tools-editor: neovim integration (3 tools) - hanzo-tools-jupyter: notebook read/edit (1 tool) - hanzo-tools-browser: Playwright browser automation (1 tool) - hanzo-tools-todo: unified todo management (1 tool) Key improvements: - All subprocess.run calls have timeout protection - All tool .call() methods are async - Namespace package pattern for mix-and-match installation - Entry point discovery via hanzo.tools - No blocking imports (all < 2s) Also updates hanzo-mcp: - Browser tool integration - Tool install/registry for dynamic tool management - Auto-background timeout reduced to 60s
99 lines
2.8 KiB
Python
99 lines
2.8 KiB
Python
"""Response truncation utilities for shell tools.
|
|
|
|
Ensures shell command output doesn't exceed token limits.
|
|
"""
|
|
|
|
import tiktoken
|
|
|
|
|
|
def estimate_tokens(text: str, model: str = "gpt-4") -> int:
|
|
"""Estimate the number of tokens in a text string.
|
|
|
|
Args:
|
|
text: The text to estimate tokens for
|
|
model: The model to use for token estimation (default: gpt-4)
|
|
|
|
Returns:
|
|
Estimated number of tokens
|
|
"""
|
|
try:
|
|
encoding = tiktoken.encoding_for_model(model)
|
|
except KeyError:
|
|
encoding = tiktoken.get_encoding("cl100k_base")
|
|
|
|
return len(encoding.encode(text))
|
|
|
|
|
|
def truncate_response(
|
|
response: str,
|
|
max_tokens: int = 20000,
|
|
truncation_message: str = "\n\n[Response truncated due to length. Please use pagination, filtering, or limit parameters to see more.]",
|
|
) -> str:
|
|
"""Truncate a response to fit within token limits.
|
|
|
|
Args:
|
|
response: The response text to truncate
|
|
max_tokens: Maximum number of tokens allowed (default: 20000)
|
|
truncation_message: Message to append when truncating
|
|
|
|
Returns:
|
|
Truncated response if needed, original response otherwise
|
|
"""
|
|
# Quick check - if response is short, no need to count tokens
|
|
if len(response) < max_tokens * 2: # Rough estimate: 1 token ≈ 2-4 chars
|
|
return response
|
|
|
|
# Estimate tokens
|
|
token_count = estimate_tokens(response)
|
|
|
|
# If within limit, return as-is
|
|
if token_count <= max_tokens:
|
|
return response
|
|
|
|
# Need to truncate
|
|
left, right = 0, len(response)
|
|
truncation_msg_tokens = estimate_tokens(truncation_message)
|
|
target_tokens = max_tokens - truncation_msg_tokens
|
|
|
|
while left < right - 1:
|
|
mid = (left + right) // 2
|
|
mid_tokens = estimate_tokens(response[:mid])
|
|
|
|
if mid_tokens <= target_tokens:
|
|
left = mid
|
|
else:
|
|
right = mid
|
|
|
|
# Find a good break point (newline or space)
|
|
truncate_at = left
|
|
for i in range(min(100, left), -1, -1):
|
|
if response[left - i] in "\n ":
|
|
truncate_at = left - i
|
|
break
|
|
|
|
return response[:truncate_at] + truncation_message
|
|
|
|
|
|
def truncate_lines(
|
|
response: str,
|
|
max_lines: int = 1000,
|
|
truncation_message: str = "\n\n[Response truncated to {max_lines} lines. Please use pagination or filtering to see more.]",
|
|
) -> str:
|
|
"""Truncate a response by number of lines.
|
|
|
|
Args:
|
|
response: The response text to truncate
|
|
max_lines: Maximum number of lines allowed (default: 1000)
|
|
truncation_message: Message template to append when truncating
|
|
|
|
Returns:
|
|
Truncated response if needed, original response otherwise
|
|
"""
|
|
lines = response.split("\n")
|
|
|
|
if len(lines) <= max_lines:
|
|
return response
|
|
|
|
truncated = "\n".join(lines[:max_lines])
|
|
return truncated + truncation_message.format(max_lines=max_lines)
|