hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: consolidate tools - 52 to 30 tools

Browse Source 
Major consolidation of hanzo-mcp tools:

Memory (9 → 1 tool):
- Unified UnifiedMemoryTool with actions: recall, create, update, delete,
  manage, facts, store, summarize, kb

Agent (10+ → 3 tools):
- Unified UnifiedAgentTool for claude, codex, gemini, grok CLI agents
- Removed duplicate critic (use reasoning.critic instead)
- Kept iching and review tools
- Removed code_auth, network, request_clarification

LLM (3 → 2 tools):
- Removed llm_manage (consolidated into llm)

Tool Management (4 → 1 tool):
- Existing unified `tool` command handles install, enable, disable, list

Essential Tools:
- Added memory, shell, open, tool to ESSENTIAL_TOOLS
- Added memory to ESSENTIAL_SYSTEM_TOOLS (always enabled)
- Updated hanzo mode with iching, review, computer

Bug fixes:
- Fixed async await warning in dag_tool.py
- Fixed duplicate critic conflict between agent and reasoning
This commit is contained in:
zeekay
2025-12-26 11:43:19 -08:00
parent 2cb7eb03fc
commit 5a1c976535
11 changed files with 762 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-mcp/hanzo_mcp/tools/__init__.py
-8
View File
@@ -228,14 +228,6 @@ def _register_system_tools(
except ImportError:
logger.debug("Stats tool not available")
# CLI tool factory (dynamic CLI tool creation)
try:
from hanzo_mcp.tools.common.cli_tool_factory import register_cli_factory_tools
register_cli_factory_tools(mcp_server)
logger.info("Registered CLI tool factory (cli_create, cli_list, cli_remove, cli_help)")
except ImportError:
logger.debug("CLI tool factory not available")
# Re-export for backward compatibility
__all__ = ["register_all_tools"]
pkg/hanzo-mcp/hanzo_mcp/tools/common/entrypoint_loader.py
+8 -17
View File
@@ -29,32 +29,23 @@ TOOLS_ENTRY_POINT_GROUP = "hanzo.tools"
# Package to tool prefix mapping (for enable/disable)
PACKAGE_TOOL_PREFIXES: dict[str, list[str]] = {
"filesystem": ["read", "write", "edit", "multi_edit", "tree", "ast", "rules", "search", "find"],
"filesystem": ["read", "write", "edit", "tree", "ast", "search", "find"],
"shell": ["dag", "ps", "zsh", "shell", "npx", "uvx", "open"],
"browser": ["browser"],
"memory": [
"recall_memories", "create_memories", "update_memories", "delete_memories",
"manage_memories", "recall_facts", "store_facts", "summarize_to_memory",
"manage_knowledge_bases"
],
"memory": ["memory"], # Unified memory tool
"todo": ["todo"],
"reasoning": ["think", "critic"],
"lsp": ["lsp"],
"refactor": ["refactor"],
"database": [
"sql_query", "sql_search", "sql_stats",
"graph_add", "graph_remove", "graph_query", "graph_search", "graph_stats"
],
"agent": [
"agent", "swarm", "dispatch_agent", "claude", "codex", "gemini", "grok",
"code_auth", "critic", "iching", "review", "clarification", "network"
],
"jupyter": ["jupyter", "notebook_read", "notebook_edit"],
"database": ["sql", "graph"], # Consolidated database tools
"agent": ["agent", "iching", "review"], # Consolidated agent tools (critic is from reasoning)
"jupyter": ["jupyter"],
"editor": ["neovim_edit", "neovim_command", "neovim_session"],
"llm": ["llm", "unified_llm", "consensus", "llm_manage"],
"llm": ["llm", "consensus"], # Removed llm_manage
"vector": ["index", "vector_index", "vector_search"],
"config": ["config", "mode"],
"mcp_tools": ["mcp", "mcp_add", "mcp_remove", "mcp_stats"],
"mcp_tools": ["mcp"], # Consolidated MCP tool
"computer": ["computer"],
}
pkg/hanzo-mcp/hanzo_mcp/tools/common/mode_loader.py
+3
View File
@@ -26,6 +26,9 @@ ESSENTIAL_SYSTEM_TOOLS: Set[str] = {
# Core system
"version", # Version info
"stats", # Statistics
# Memory (always available)
"memory", # Unified memory operations
}
pkg/hanzo-mcp/hanzo_mcp/tools/common/personality.py
+7 -2
View File
@@ -102,13 +102,17 @@ ESSENTIAL_TOOLS = [
# File operations
"read", "write", "edit", "tree",
# Shell
"dag", "zsh",
"dag", "zsh", "shell", "open",
# Memory
"memory",
# Reasoning
"think", "critic",
# LLM infrastructure (always available)
"llm", "consensus",
# Configuration
"config", "mode",
# Tool management
"tool",
]
# Common tool sets for reuse
@@ -153,8 +157,9 @@ def _register_builtin_personalities() -> None:
description="Balanced productivity and quality",
philosophy="The Zen of Model Context Protocol.",
tools=list(set(ESSENTIAL_TOOLS + [
"agent", "todo", "rules", "browser",
"agent", "todo", "browser", "computer",
"search", "find", "ast", "jupyter", "refactor", "lsp",
"iching", "review", # Agent sub-tools
] + BUILD_TOOLS)),
environment={"HANZO_MODE": "zen"},
),
pkg/hanzo-mcp/uv.lock
Generated
+1 -1
View File
@@ -573,7 +573,7 @@ wheels = [
[[package]]
name = "hanzo-mcp"
version = "0.10.19"
version = "0.10.20"
source = { editable = "." }
dependencies = [
{ name = "aiofiles" },
pkg/hanzo-tools-agent/hanzo_tools/agent/__init__.py
+21 -58
View File
@@ -1,10 +1,9 @@
"""Agent tools for Hanzo AI.
This package provides agent orchestration and multi-agent tools:
- agent: Agent spawning and orchestration
- swarm: Multi-agent swarm coordination
- CLI agents: Claude, Codex, Gemini, Grok CLI tools
- critic: Critical analysis and review tool
Provides consolidated agent tools:
- agent: Unified agent runner (claude, codex, gemini, grok)
- critic: Critical analysis tool
- iching: I Ching wisdom for decisions
- review: Code review tool
Install:
@@ -15,70 +14,42 @@ Usage:
# Register with MCP server
register_tools(mcp_server)
Optional dependencies:
pip install hanzo-tools-agent[full] # For LLM provider support
"""
import logging
logger = logging.getLogger(__name__)
# Core tools that don't have external dependencies
from .critic_tool import CriticTool
# Core consolidated tools
from .unified_agent_tool import UnifiedAgentTool
from .iching_tool import IChingTool
from .review_tool import ReviewTool
from .network_tool import NetworkTool
from .grok_cli_tool import GrokCLITool
# CLI agent tools
# Legacy imports for backwards compatibility
from .cli_agent_base import CLIAgentBase
from .code_auth_tool import CodeAuthTool
from .codex_cli_tool import CodexCLITool
from .claude_cli_tool import ClaudeCLITool
from .codex_cli_tool import CodexCLITool
from .gemini_cli_tool import GeminiCLITool
from .clarification_tool import ClarificationTool
from .grok_cli_tool import GrokCLITool
from .critic_tool import CriticTool # Legacy - use reasoning.critic instead
# Tools with optional dependencies - imported lazily
_optional_tools = []
try:
from .agent_tool import AgentTool
_optional_tools.append(AgentTool)
except ImportError as e:
logger.debug(f"AgentTool not available: {e}")
try:
from .swarm_tool import SwarmTool
_optional_tools.append(SwarmTool)
except ImportError as e:
logger.debug(f"SwarmTool not available: {e}")
# Export list for tool discovery
# Export list for tool discovery - consolidated to 3 tools
# Note: critic is from hanzo-tools-reasoning, not here
TOOLS = [
CriticTool,
IChingTool,
ReviewTool,
ClarificationTool,
NetworkTool,
CodeAuthTool,
ClaudeCLITool,
CodexCLITool,
GeminiCLITool,
GrokCLITool,
] + _optional_tools
UnifiedAgentTool, # Unified agent runner
IChingTool, # I Ching wisdom
ReviewTool, # Code review
]
__all__ = [
"register_tools",
"TOOLS",
# Primary tools
"UnifiedAgentTool",
"CriticTool",
"IChingTool",
"ReviewTool",
"ClarificationTool",
"NetworkTool",
"CodeAuthTool",
# Legacy (for backwards compatibility)
"CLIAgentBase",
"ClaudeCLITool",
"CodexCLITool",
@@ -88,15 +59,7 @@ __all__ = [
def register_tools(mcp_server, enabled_tools: dict[str, bool] | None = None):
"""Register all agent tools with the MCP server.
Args:
mcp_server: FastMCP server instance
enabled_tools: Dict of tool_name -> enabled state
Returns:
List of registered tool instances
"""
"""Register agent tools with the MCP server."""
from hanzo_tools.core import ToolRegistry
enabled = enabled_tools or {}
@@ -105,7 +68,7 @@ def register_tools(mcp_server, enabled_tools: dict[str, bool] | None = None):
for tool_class in TOOLS:
tool_name = tool_class.name if hasattr(tool_class, "name") else tool_class.__name__.lower()
if enabled.get(tool_name, True): # Enabled by default
if enabled.get(tool_name, True):
try:
tool = tool_class()
ToolRegistry.register_tool(mcp_server, tool)
pkg/hanzo-tools-agent/hanzo_tools/agent/unified_agent_tool.py
+232
View File
@@ -0,0 +1,232 @@
"""Unified agent tool - run any agent by name.
Single tool that dispatches to installed CLI agents (claude, codex, gemini, grok)
and orchestrates multi-agent workflows.
"""
from typing import Optional, List, Dict, Any, Literal, final, override, Annotated
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
Action = Annotated[
Literal[
"run", # Run a specific agent
"list", # List available agents
"status", # Check agent status
],
Field(description="Agent action to perform"),
]
@final
class UnifiedAgentTool(BaseTool):
"""Unified agent tool for running CLI agents.
Dispatches to: claude, codex, gemini, grok or custom agents.
"""
name = "agent"
# Available agent configurations
AGENTS = {
"claude": {
"command": "claude",
"description": "Anthropic Claude Code CLI",
"check": ["claude", "--version"],
},
"codex": {
"command": "codex",
"description": "OpenAI Codex CLI",
"check": ["codex", "--version"],
},
"gemini": {
"command": "gemini",
"description": "Google Gemini CLI",
"check": ["gemini", "--version"],
},
"grok": {
"command": "grok",
"description": "xAI Grok CLI",
"check": ["grok", "--version"],
},
}
@property
@override
def description(self) -> str:
return """Run AI agents by name.
Actions:
- run: Execute an agent with a prompt
- list: List available agents
- status: Check if an agent is available
Agents: claude, codex, gemini, grok
Examples:
agent run --name claude --prompt "Explain this code"
agent run --name codex --prompt "Write a function" --cwd /path/to/project
agent list
agent status --name claude
"""
@override
@auto_timeout("agent")
async def call(
self,
ctx: MCPContext,
action: str = "list",
name: Optional[str] = None,
prompt: Optional[str] = None,
cwd: Optional[str] = None,
args: Optional[List[str]] = None,
timeout: int = 300,
**kwargs,
) -> str:
"""Execute agent operation."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
if action == "list":
return self._list_agents()
elif action == "status":
return await self._check_status(name)
elif action == "run":
return await self._run_agent(name, prompt, cwd, args, timeout)
else:
return f"Unknown action: {action}. Use: run, list, status"
def _list_agents(self) -> str:
"""List available agents."""
lines = ["Available agents:"]
for name, config in self.AGENTS.items():
lines.append(f" • {name}: {config['description']}")
lines.append("")
lines.append("Usage: agent run --name <agent> --prompt 'your prompt'")
return "\n".join(lines)
async def _check_status(self, name: Optional[str]) -> str:
"""Check if an agent is available."""
import asyncio
if not name:
# Check all agents
results = []
for agent_name, config in self.AGENTS.items():
available = await self._is_available(config["check"])
status = "✓ available" if available else "✗ not found"
results.append(f" {agent_name}: {status}")
return "Agent status:\n" + "\n".join(results)
if name not in self.AGENTS:
return f"Unknown agent: {name}. Available: {', '.join(self.AGENTS.keys())}"
config = self.AGENTS[name]
available = await self._is_available(config["check"])
if available:
return f"✓ {name} is available"
return f"✗ {name} is not installed or not in PATH"
async def _is_available(self, check_cmd: List[str]) -> bool:
"""Check if a command is available."""
import asyncio
try:
proc = await asyncio.create_subprocess_exec(
*check_cmd,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await asyncio.wait_for(proc.wait(), timeout=5)
return proc.returncode == 0
except Exception:
return False
async def _run_agent(
self,
name: Optional[str],
prompt: Optional[str],
cwd: Optional[str],
args: Optional[List[str]],
timeout: int,
) -> str:
"""Run an agent with a prompt."""
import asyncio
import os
if not name:
return "Error: name required for run action"
if not prompt:
return "Error: prompt required for run action"
if name not in self.AGENTS:
return f"Unknown agent: {name}. Available: {', '.join(self.AGENTS.keys())}"
config = self.AGENTS[name]
command = config["command"]
# Build command
cmd_args = [command, prompt]
if args:
cmd_args.extend(args)
try:
proc = await asyncio.create_subprocess_exec(
*cmd_args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd or os.getcwd(),
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout,
)
output = stdout.decode("utf-8", errors="replace")
if proc.returncode != 0:
err = stderr.decode("utf-8", errors="replace")
return f"Agent {name} failed (exit {proc.returncode}):\n{output}\n{err}"
return f"[{name}] {output}"
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
return f"Agent {name} timed out after {timeout}s"
except FileNotFoundError:
return f"Agent {name} not found. Install it or check PATH."
except Exception as e:
return f"Error running {name}: {e}"
def register(self, mcp_server: FastMCP) -> None:
"""Register with MCP server."""
tool_instance = self
@mcp_server.tool()
async def agent(
action: Action = "list",
name: Annotated[Optional[str], Field(description="Agent name: claude, codex, gemini, grok")] = None,
prompt: Annotated[Optional[str], Field(description="Prompt for the agent")] = None,
cwd: Annotated[Optional[str], Field(description="Working directory")] = None,
args: Annotated[Optional[List[str]], Field(description="Additional arguments")] = None,
timeout: Annotated[int, Field(description="Timeout in seconds")] = 300,
ctx: MCPContext = None,
) -> str:
"""Run AI agents: claude, codex, gemini, grok."""
return await tool_instance.call(
ctx,
action=action,
name=name,
prompt=prompt,
cwd=cwd,
args=args,
timeout=timeout,
)
pkg/hanzo-tools-llm/hanzo_tools/llm/__init__.py
-8
View File
@@ -34,13 +34,6 @@ except ImportError as e:
logger.debug(f"ConsensusTool not available: {e}")
ConsensusTool = None
# LLMManageTool kept for backwards compatibility but not exported to TOOLS
try:
from .llm_manage import LLMManageTool
except ImportError as e:
logger.debug(f"LLMManageTool not available: {e}")
LLMManageTool = None
TOOLS = _tools
LLM_AVAILABLE = len(_tools) > 0
@@ -50,7 +43,6 @@ __all__ = [
"LLMTool",
"UnifiedLLMTool",
"ConsensusTool",
"LLMManageTool",
"register_tools",
]
pkg/hanzo-tools-memory/hanzo_tools/memory/__init__.py
+21 -49
View File
@@ -1,20 +1,23 @@
"""Memory and knowledge tools for Hanzo MCP.
This package provides tools for managing memories and knowledge bases:
- RecallMemoriesTool: Search and retrieve memories
- CreateMemoriesTool: Store new memories
- UpdateMemoriesTool: Update existing memories
- DeleteMemoriesTool: Remove memories
- ManageMemoriesTool: Atomic memory operations
- RecallFactsTool: Search knowledge bases
- StoreFactsTool: Store facts in knowledge bases
- SummarizeToMemoryTool: Summarize and store information
- ManageKnowledgeBasesTool: Create and manage knowledge bases
Provides a unified `memory` tool for all memory and knowledge operations:
- recall: Search and retrieve memories
- create: Store new memories
- update: Update existing memories
- delete: Remove memories
- manage: Atomic operations
- facts: Recall from knowledge bases
- store: Store facts
- summarize: Summarize and store
- kb: Manage knowledge bases
"""
from mcp.server import FastMCP
from hanzo_tools.core import BaseTool, ToolRegistry
from hanzo_tools.memory.unified_memory_tool import UnifiedMemoryTool
# Legacy imports for backwards compatibility
from hanzo_tools.memory.memory_tools import (
MEMORY_TOOL_CLASSES,
CreateMemoriesTool,
@@ -32,13 +35,14 @@ from hanzo_tools.memory.knowledge_tools import (
)
__all__ = [
# Memory tools
# Primary unified tool
"UnifiedMemoryTool",
# Legacy tools (for backwards compatibility)
"RecallMemoriesTool",
"CreateMemoriesTool",
"UpdateMemoriesTool",
"DeleteMemoriesTool",
"ManageMemoriesTool",
# Knowledge tools
"RecallFactsTool",
"StoreFactsTool",
"SummarizeToMemoryTool",
@@ -49,26 +53,14 @@ __all__ = [
"TOOLS",
]
# All tool classes for the TOOLS entry point
ALL_TOOL_CLASSES = MEMORY_TOOL_CLASSES + KNOWLEDGE_TOOL_CLASSES
def get_memory_tools(
user_id: str = "default",
project_id: str = "default",
**kwargs,
) -> list[BaseTool]:
"""Create instances of all memory and knowledge tools.
Args:
user_id: User ID for memory operations
project_id: Project ID for memory operations
**kwargs: Additional configuration
Returns:
List of tool instances
"""
return [cls(user_id=user_id, project_id=project_id, **kwargs) for cls in ALL_TOOL_CLASSES]
"""Create unified memory tool instance."""
return [UnifiedMemoryTool(user_id=user_id, project_id=project_id, **kwargs)]
def register_memory_tools(
@@ -77,31 +69,11 @@ def register_memory_tools(
project_id: str = "default",
**kwargs,
) -> list[BaseTool]:
"""Register all memory and knowledge tools with the MCP server.
Args:
mcp_server: The FastMCP server instance
user_id: User ID for memory operations
project_id: Project ID for memory operations
**kwargs: Additional configuration
Returns:
List of registered tools
"""
"""Register unified memory tool with the MCP server."""
tools = get_memory_tools(user_id=user_id, project_id=project_id, **kwargs)
ToolRegistry.register_tools(mcp_server, tools)
return tools
# TOOLS list for entry point discovery
TOOLS = [
RecallMemoriesTool,
CreateMemoriesTool,
UpdateMemoriesTool,
DeleteMemoriesTool,
ManageMemoriesTool,
RecallFactsTool,
StoreFactsTool,
SummarizeToMemoryTool,
ManageKnowledgeBasesTool,
]
# TOOLS list for entry point discovery - single unified tool
TOOLS = [UnifiedMemoryTool]
pkg/hanzo-tools-memory/hanzo_tools/memory/unified_memory_tool.py
+468
View File
@@ -0,0 +1,468 @@
"""Unified memory tool - consolidates all memory operations.
Single tool with action parameter replaces 9 separate memory tools.
"""
from typing import TYPE_CHECKING, Optional, List, Dict, Any, Literal, final, override, Annotated
from pydantic import Field
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
if TYPE_CHECKING:
from hanzo_memory.services.memory import MemoryService
# Lazy loading
MEMORY_AVAILABLE: Optional[bool] = None
_memory_service: Optional["MemoryService"] = None
def _check_memory_available() -> bool:
"""Check if hanzo-memory is available."""
global MEMORY_AVAILABLE
if MEMORY_AVAILABLE is None:
try:
import hanzo_memory # noqa: F401
MEMORY_AVAILABLE = True
except ImportError:
MEMORY_AVAILABLE = False
return MEMORY_AVAILABLE
def _get_lazy_memory_service() -> "MemoryService":
"""Get memory service lazily."""
global _memory_service
if _memory_service is None:
if not _check_memory_available():
raise ImportError("hanzo-memory required. Install: pip install hanzo-memory")
from hanzo_memory.services.memory import get_memory_service
_memory_service = get_memory_service()
return _memory_service
Action = Annotated[
Literal[
"recall", # Search memories
"create", # Store new memories
"update", # Update existing memories
"delete", # Delete memories
"manage", # Atomic create/update/delete
"facts", # Recall facts from knowledge bases
"store", # Store facts in knowledge bases
"summarize", # Summarize and store
"kb", # Manage knowledge bases
"list", # List memories/knowledge bases
],
Field(description="Memory action to perform"),
]
@final
class UnifiedMemoryTool(BaseTool):
"""Unified memory tool for all memory operations.
Consolidates: recall_memories, create_memories, update_memories,
delete_memories, manage_memories, recall_facts, store_facts,
summarize_to_memory, manage_knowledge_bases
"""
name = "memory"
def __init__(self, user_id: str = "default", project_id: str = "default", **kwargs):
"""Initialize memory tool."""
self.user_id = user_id
self.project_id = project_id
self._service: Optional["MemoryService"] = None
@property
def service(self) -> "MemoryService":
"""Get memory service lazily."""
if self._service is None:
self._service = _get_lazy_memory_service()
return self._service
@property
@override
def description(self) -> str:
return """Unified memory management tool.
Actions:
- recall: Search and retrieve memories
- create: Store new memories
- update: Update existing memories
- delete: Remove memories
- manage: Atomic operations (create + update + delete)
- facts: Recall facts from knowledge bases
- store: Store facts in knowledge bases
- summarize: Summarize information and store
- kb: Create/list/manage knowledge bases
- list: List all memories or knowledge bases
Examples:
memory recall --query "user preferences"
memory create --content "User prefers dark mode" --tags ["preferences"]
memory update --id mem_123 --content "Updated content"
memory delete --id mem_123
memory facts --query "coding standards" --kb "project_docs"
memory store --fact "Python uses 4-space indentation" --kb "coding"
memory kb --action create --name "project_notes"
memory list --type memories
"""
@override
@auto_timeout("memory")
async def call(
self,
ctx: MCPContext,
action: str = "list",
# Common params
query: Optional[str] = None,
queries: Optional[List[str]] = None,
content: Optional[str] = None,
id: Optional[str] = None,
ids: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
scope: str = "session",
limit: int = 10,
# Knowledge base params
kb: Optional[str] = None,
fact: Optional[str] = None,
facts: Optional[List[str]] = None,
# Manage params
create_list: Optional[List[Dict]] = None,
update_list: Optional[List[Dict]] = None,
delete_ids: Optional[List[str]] = None,
# KB management
kb_action: Optional[str] = None,
name: Optional[str] = None,
# List params
list_type: str = "memories",
**kwargs,
) -> str:
"""Execute memory operation."""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
try:
if action == "recall":
return await self._recall(queries or ([query] if query else []), scope, limit)
elif action == "create":
return await self._create(content, tags, metadata, scope)
elif action == "update":
return await self._update(id, content, tags, metadata)
elif action == "delete":
return await self._delete(ids or ([id] if id else []))
elif action == "manage":
return await self._manage(create_list, update_list, delete_ids)
elif action == "facts":
return await self._recall_facts(queries or ([query] if query else []), kb, limit)
elif action == "store":
return await self._store_facts(facts or ([fact] if fact else []), kb, metadata)
elif action == "summarize":
return await self._summarize(content, tags, scope)
elif action == "kb":
return await self._manage_kb(kb_action or "list", name)
elif action == "list":
return await self._list(list_type, scope, limit)
else:
return f"Unknown action: {action}. Use: recall, create, update, delete, manage, facts, store, summarize, kb, list"
except ImportError as e:
return f"Memory service not available: {e}"
except Exception as e:
return f"Memory operation failed: {e}"
async def _recall(self, queries: List[str], scope: str, limit: int) -> str:
"""Search and retrieve memories."""
if not queries:
return "Error: query or queries required for recall"
results = []
for q in queries:
memories = await self.service.recall(
query=q,
user_id=self.user_id,
project_id=self.project_id,
scope=scope,
limit=limit,
)
results.extend(memories)
if not results:
return "No memories found matching the query."
lines = [f"Found {len(results)} memories:"]
for mem in results[:limit]:
lines.append(f" [{mem.id}] {mem.content[:100]}...")
if mem.tags:
lines.append(f" Tags: {', '.join(mem.tags)}")
return "\n".join(lines)
async def _create(
self,
content: Optional[str],
tags: Optional[List[str]],
metadata: Optional[Dict],
scope: str,
) -> str:
"""Store new memory."""
if not content:
return "Error: content required for create"
memory = await self.service.create(
content=content,
user_id=self.user_id,
project_id=self.project_id,
tags=tags or [],
metadata=metadata or {},
scope=scope,
)
return f"Created memory: {memory.id}"
async def _update(
self,
id: Optional[str],
content: Optional[str],
tags: Optional[List[str]],
metadata: Optional[Dict],
) -> str:
"""Update existing memory."""
if not id:
return "Error: id required for update"
await self.service.update(
memory_id=id,
content=content,
tags=tags,
metadata=metadata,
)
return f"Updated memory: {id}"
async def _delete(self, ids: List[str]) -> str:
"""Delete memories."""
if not ids:
return "Error: id or ids required for delete"
for mid in ids:
await self.service.delete(memory_id=mid)
return f"Deleted {len(ids)} memories"
async def _manage(
self,
create_list: Optional[List[Dict]],
update_list: Optional[List[Dict]],
delete_ids: Optional[List[str]],
) -> str:
"""Atomic memory operations."""
results = []
if create_list:
for item in create_list:
mem = await self.service.create(
content=item["content"],
user_id=self.user_id,
project_id=self.project_id,
tags=item.get("tags", []),
metadata=item.get("metadata", {}),
)
results.append(f"Created: {mem.id}")
if update_list:
for item in update_list:
await self.service.update(
memory_id=item["id"],
content=item.get("content"),
tags=item.get("tags"),
metadata=item.get("metadata"),
)
results.append(f"Updated: {item['id']}")
if delete_ids:
for mid in delete_ids:
await self.service.delete(memory_id=mid)
results.append(f"Deleted: {mid}")
if not results:
return "No operations specified. Provide create_list, update_list, or delete_ids."
return "\n".join(results)
async def _recall_facts(self, queries: List[str], kb: Optional[str], limit: int) -> str:
"""Search knowledge bases."""
if not queries:
return "Error: query required for facts"
try:
from hanzo_memory.services.knowledge import get_knowledge_service
ks = get_knowledge_service()
results = []
for q in queries:
facts = await ks.recall(query=q, knowledge_base=kb, limit=limit)
results.extend(facts)
if not results:
return "No facts found."
lines = [f"Found {len(results)} facts:"]
for fact in results[:limit]:
lines.append(f" • {fact.content[:100]}...")
return "\n".join(lines)
except ImportError:
return "Knowledge service not available"
async def _store_facts(
self,
facts: List[str],
kb: Optional[str],
metadata: Optional[Dict],
) -> str:
"""Store facts in knowledge base."""
if not facts:
return "Error: fact or facts required"
try:
from hanzo_memory.services.knowledge import get_knowledge_service
ks = get_knowledge_service()
stored = []
for fact in facts:
result = await ks.store(
content=fact,
knowledge_base=kb or "default",
metadata=metadata or {},
)
stored.append(result.id)
return f"Stored {len(stored)} facts in '{kb or 'default'}'"
except ImportError:
return "Knowledge service not available"
async def _summarize(
self,
content: Optional[str],
tags: Optional[List[str]],
scope: str,
) -> str:
"""Summarize and store."""
if not content:
return "Error: content required for summarize"
# For now, just store as-is; could integrate with LLM for actual summarization
memory = await self.service.create(
content=f"[Summary] {content}",
user_id=self.user_id,
project_id=self.project_id,
tags=(tags or []) + ["summary"],
scope=scope,
)
return f"Stored summary: {memory.id}"
async def _manage_kb(self, action: str, name: Optional[str]) -> str:
"""Manage knowledge bases."""
try:
from hanzo_memory.services.knowledge import get_knowledge_service
ks = get_knowledge_service()
if action == "list":
kbs = await ks.list_knowledge_bases()
if not kbs:
return "No knowledge bases found."
lines = ["Knowledge bases:"]
for kb in kbs:
lines.append(f" • {kb.name}: {kb.description or 'No description'}")
return "\n".join(lines)
elif action == "create":
if not name:
return "Error: name required for kb create"
await ks.create_knowledge_base(name=name)
return f"Created knowledge base: {name}"
elif action == "delete":
if not name:
return "Error: name required for kb delete"
await ks.delete_knowledge_base(name=name)
return f"Deleted knowledge base: {name}"
else:
return f"Unknown kb action: {action}. Use: list, create, delete"
except ImportError:
return "Knowledge service not available"
async def _list(self, list_type: str, scope: str, limit: int) -> str:
"""List memories or knowledge bases."""
if list_type == "kb":
return await self._manage_kb("list", None)
# List memories
memories = await self.service.list(
user_id=self.user_id,
project_id=self.project_id,
scope=scope,
limit=limit,
)
if not memories:
return "No memories found."
lines = [f"Memories ({len(memories)}):"]
for mem in memories:
lines.append(f" [{mem.id}] {mem.content[:60]}...")
return "\n".join(lines)
def register(self, mcp_server: FastMCP) -> None:
"""Register with MCP server."""
tool_instance = self
@mcp_server.tool()
async def memory(
action: Action = "list",
query: Annotated[Optional[str], Field(description="Search query")] = None,
queries: Annotated[Optional[List[str]], Field(description="Multiple queries")] = None,
content: Annotated[Optional[str], Field(description="Memory content")] = None,
id: Annotated[Optional[str], Field(description="Memory ID")] = None,
ids: Annotated[Optional[List[str]], Field(description="Multiple IDs")] = None,
tags: Annotated[Optional[List[str]], Field(description="Tags")] = None,
metadata: Annotated[Optional[Dict], Field(description="Metadata")] = None,
scope: Annotated[str, Field(description="Scope: session, project, global")] = "session",
limit: Annotated[int, Field(description="Max results")] = 10,
kb: Annotated[Optional[str], Field(description="Knowledge base name")] = None,
fact: Annotated[Optional[str], Field(description="Fact to store")] = None,
facts: Annotated[Optional[List[str]], Field(description="Multiple facts")] = None,
create_list: Annotated[Optional[List[Dict]], Field(description="Items to create")] = None,
update_list: Annotated[Optional[List[Dict]], Field(description="Items to update")] = None,
delete_ids: Annotated[Optional[List[str]], Field(description="IDs to delete")] = None,
kb_action: Annotated[Optional[str], Field(description="KB action: list, create, delete")] = None,
name: Annotated[Optional[str], Field(description="KB name")] = None,
list_type: Annotated[str, Field(description="List type: memories, kb")] = "memories",
ctx: MCPContext = None,
) -> str:
"""Unified memory management: recall, create, update, delete, facts, kb."""
return await tool_instance.call(
ctx,
action=action,
query=query,
queries=queries,
content=content,
id=id,
ids=ids,
tags=tags,
metadata=metadata,
scope=scope,
limit=limit,
kb=kb,
fact=fact,
facts=facts,
create_list=create_list,
update_list=update_list,
delete_ids=delete_ids,
kb_action=kb_action,
name=name,
list_type=list_type,
)
pkg/hanzo-tools-shell/hanzo_tools/shell/dag_tool.py
+1 -1
View File
@@ -423,7 +423,7 @@ backgrounded. Use ps tool to monitor: ps --logs <id>, ps --kill <id>"""
) -> str:
"""Execute commands with DAG semantics."""
tool_ctx = create_tool_context(ctx)
tool_ctx.set_tool_info(self.name)
await tool_ctx.set_tool_info(self.name)
shell = shell or self.default_shell
results: List[DagResult] = []