RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tool dedup

This commit is contained in:
Shannon Sands 2026-02-02 15:28:10 +10:00
parent 8dccd6569e
commit 4939130485
18 changed files with 376 additions and 4288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
atropos/api/tool_executor_server.py
View File

@ -19,11 +19,7 @@ from fastapi import FastAPI, Header, HTTPException, status
from pydantic import BaseModel, Field
from ..slots import SlotPool, SlotPoolConfig
from ..tools import BashTool, ImageGenerateTool, ReadFileTool, ToolRegistry, WriteFileTool
from ..tools.mixture_of_agents_tool import MixtureOfAgentsTool
from ..tools.terminal_tool import TerminalTool
from ..tools.vision_tools import VisionAnalyzeTool
from ..tools.web_tools import WebCrawlTool, WebExtractTool, WebSearchTool
from ..tools import ToolRegistry, build_tool_registry
from ..tools.base import (
ArtifactArchiveRequestPayload,
ArtifactArchiveResponsePayload,
@ -120,17 +116,12 @@ def _check_auth(cfg: ToolExecutorServerConfig, authorization: Optional[str]) ->
async def _startup() -> None:
cfg = ToolExecutorServerConfig.from_env()
tools = ToolRegistry()
tools.register(BashTool())
tools.register(TerminalTool())
tools.register(ReadFileTool())
tools.register(WriteFileTool())
tools.register(ImageGenerateTool())
tools.register(WebSearchTool())
tools.register(WebExtractTool())
tools.register(WebCrawlTool())
tools.register(VisionAnalyzeTool())
tools.register(MixtureOfAgentsTool())
# Default to Atropos "full" tool surface: sandbox + external (if tool_server_url provided).
tools: ToolRegistry = build_tool_registry(
enabled_toolsets=["full"],
disabled_toolsets=None,
tool_server_url=cfg.tool_server_url,
)
pool = SlotPool(
SlotPoolConfig(

39
atropos/api/tool_server.py
View File

@ -19,12 +19,8 @@ from pathlib import Path
from fastapi import FastAPI, Header, HTTPException, status
from pydantic import BaseModel, Field
from ..tools import ToolRegistry
from ..tools import ToolRegistry, build_tool_registry
from ..tools.base import ToolResultPayload, ToolServerExecuteRequest
from ..tools.image_generation_tool import ImageGenerateTool
from ..tools.mixture_of_agents_tool import MixtureOfAgentsTool
from ..tools.vision_tools import VisionAnalyzeTool
from ..tools.web_tools import WebCrawlTool, WebExtractTool, WebSearchTool
class ToolServerConfig(BaseModel):
@ -63,21 +59,13 @@ async def root() -> Dict[str, str]:
async def _startup() -> None:
cfg = ToolServerConfig.from_env()
tools = ToolRegistry()
for tool in [
ImageGenerateTool(),
WebSearchTool(),
WebExtractTool(),
WebCrawlTool(),
VisionAnalyzeTool(),
MixtureOfAgentsTool(),
]:
ok, reason = tool.is_available()
if ok:
tools.register(tool)
else:
# Keep startup resilient when optional deps/keys are missing.
print(f"[ToolServer] Skipping tool '{tool.name}': {reason}")
# External-only registry. It will only include tools that are enabled by toolsets and
# whose Hermes requirements/keys are satisfied in this process.
tools: ToolRegistry = build_tool_registry(
enabled_toolsets=["all"],
disabled_toolsets=["terminal", "sandbox", "filesystem", "terminal_stateful", "default"],
tool_server_url="enabled",
)
app.state.cfg = cfg
app.state.tools = tools
@ -129,13 +117,16 @@ async def execute_tool(
async with sem:
try:
kwargs = dict(req.tool.arguments)
# Some external tools need access to the trajectory/workspace context (e.g. fetching sandbox artifacts).
if req.trajectory_id and "trajectory_id" in inspect.signature(tool.execute).parameters:
sig = inspect.signature(tool.execute).parameters
# Some tools can benefit from extra context.
if req.trajectory_id and "trajectory_id" in sig:
kwargs["trajectory_id"] = req.trajectory_id
if req.slot_id and "slot_id" in inspect.signature(tool.execute).parameters:
if req.slot_id and "slot_id" in sig:
kwargs["slot_id"] = req.slot_id
if req.container_addr and "container_addr" in inspect.signature(tool.execute).parameters:
if req.container_addr and "container_addr" in sig:
kwargs["container_addr"] = req.container_addr
if "task_id" in sig:
kwargs["task_id"] = req.trajectory_id
result = await tool.execute(**kwargs)
except Exception as e:
return ToolResultPayload(

51
atropos/envs/agent_env.py
View File

@ -20,15 +20,7 @@ from atroposlib.envs.base import APIServerConfig, BaseEnv, BaseEnvConfig, Item,
from ..agent import AgentConfig, AtroposAgent
from ..slots import SlotPool, SlotPoolConfig
from ..tools import BashTool, ReadFileTool, ToolRegistry, WriteFileTool
from ..tools.image_generation_tool import ImageGenerateTool
from ..tools.mixture_of_agents_tool import MixtureOfAgentsTool
from ..tools.terminal_tool import TerminalTool
from ..tools.terminal_stateful_tool import TerminalStatefulTool
from ..tools.tmux_tool import TmuxTool
from ..tools.toolsets import resolve_multiple_toolsets
from ..tools.vision_tools import VisionAnalyzeTool
from ..tools.web_tools import WebCrawlTool, WebExtractTool, WebSearchTool
from ..tools import ToolRegistry, build_tool_registry
from ..tools.tool_executor import ToolExecutor, ToolExecutorConfig
@ -110,42 +102,11 @@ class AgentEnv(BaseEnv, ABC, Generic[AgentEnvConfigT]):
self._tool_server_inprocess: bool = False
def build_tools(self) -> ToolRegistry:
available_tools = [
BashTool(),
TerminalTool(),
TerminalStatefulTool(),
TmuxTool(),
ReadFileTool(),
WriteFileTool(),
ImageGenerateTool(),
WebSearchTool(),
WebExtractTool(),
WebCrawlTool(),
VisionAnalyzeTool(),
MixtureOfAgentsTool(),
]
tool_by_name = {t.name: t for t in available_tools}
enabled_toolsets = self.config.enabled_toolsets or ["default"]
selected = set(resolve_multiple_toolsets(enabled_toolsets))
if self.config.disabled_toolsets:
selected -= set(resolve_multiple_toolsets(self.config.disabled_toolsets))
tools = ToolRegistry()
for name in sorted(selected):
tool = tool_by_name.get(name)
if tool is None:
continue
# External tools require a ToolServer URL; avoid advertising broken tools.
if tool.schema.external and not self.config.tool_server_url:
continue
ok, _reason = tool.is_available()
if not ok:
continue
tools.register(tool)
return tools
return build_tool_registry(
enabled_toolsets=self.config.enabled_toolsets or ["default"],
disabled_toolsets=self.config.disabled_toolsets or None,
tool_server_url=self.config.tool_server_url,
)
@abstractmethod
def build_task(self, item: Item) -> str:

15
atropos/tools/__init__.py
View File

@ -5,14 +5,10 @@ Provides base Tool class and common tool implementations.
"""
from .base import Tool, ToolCall, ToolRegistry, ToolResult, ToolSchema
from .basic_tools import BashTool, ReadFileTool, WriteFileTool
from .image_generation_tool import ImageGenerateTool
from .mixture_of_agents_tool import MixtureOfAgentsTool
from .terminal_tool import TerminalTool
from .build_registry import build_tool_registry
from .sandbox_stubs import BashTool, ReadFileTool, TerminalTool, WriteFileTool
from .terminal_stateful_tool import TerminalStatefulTool
from .tmux_tool import TmuxTool
from .vision_tools import VisionAnalyzeTool
from .web_tools import WebCrawlTool, WebExtractTool, WebSearchTool
__all__ = [
"Tool",
@ -23,13 +19,8 @@ __all__ = [
"BashTool",
"ReadFileTool",
"WriteFileTool",
"ImageGenerateTool",
"TerminalTool",
"TerminalStatefulTool",
"TmuxTool",
"WebSearchTool",
"WebExtractTool",
"WebCrawlTool",
"VisionAnalyzeTool",
"MixtureOfAgentsTool",
"build_tool_registry",
]

283
atropos/tools/basic_tools.py
View File

@ -1,283 +0,0 @@
"""
Basic tool implementations for atropos-agent.
These tools provide simple sandbox operations:
- BashTool: Execute shell commands
- ReadFileTool: Read file contents
- WriteFileTool: Write content to files
For PoC, these run via subprocess in the local environment.
Production usage should use proper sandbox isolation.
"""
import asyncio
import os
from pathlib import Path
from typing import Optional
from .base import Tool, ToolResult, ToolSchema
class BashTool(Tool):
"""
Execute bash commands in a sandboxed environment.
TODO: Nomad slot execution
"""
def __init__(
self,
working_dir: Optional[str] = None,
timeout: float = 30.0,
max_output_size: int = 10000,
):
self.working_dir = working_dir or os.getcwd()
self.timeout = timeout
self.max_output_size = max_output_size
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="bash",
description="Execute a bash command and return stdout/stderr. Use for running shell commands, scripts, or system operations.",
parameters={
"command": {
"type": "string",
"description": "The bash command to execute",
},
},
required=["command"],
)
async def execute(self, command: str) -> ToolResult:
"""Execute a bash command."""
try:
process = await asyncio.create_subprocess_shell(
command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=self.working_dir,
)
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(),
timeout=self.timeout,
)
except asyncio.TimeoutError:
process.kill()
await process.wait()
return ToolResult(
success=False,
error=f"Command timed out after {self.timeout}s",
metadata={"exit_code": -1, "timeout": True},
)
stdout_str = stdout.decode("utf-8", errors="replace")
stderr_str = stderr.decode("utf-8", errors="replace")
# Truncate if too long
if len(stdout_str) > self.max_output_size:
stdout_str = stdout_str[:self.max_output_size] + "\n... (output truncated)"
if len(stderr_str) > self.max_output_size:
stderr_str = stderr_str[:self.max_output_size] + "\n... (output truncated)"
exit_code = process.returncode
success = exit_code == 0
output = stdout_str
if stderr_str:
output = f"{stdout_str}\n[stderr]\n{stderr_str}" if stdout_str else stderr_str
return ToolResult(
success=success,
output=output.strip(),
error="" if success else f"Exit code: {exit_code}",
metadata={"exit_code": exit_code},
)
except Exception as e:
return ToolResult(
success=False,
error=f"Failed to execute command: {str(e)}",
)
class ReadFileTool(Tool):
"""Read the contents of a file."""
def __init__(
self,
working_dir: Optional[str] = None,
max_file_size: int = 100000,
):
self.working_dir = Path(working_dir) if working_dir else Path.cwd()
self.max_file_size = max_file_size
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="read_file",
description="Read the contents of a file at the given path.",
parameters={
"path": {
"type": "string",
"description": "Path to the file (relative to working directory)",
},
},
required=["path"],
)
async def execute(self, path: str) -> ToolResult:
"""Read a file's contents."""
try:
file_path = self.working_dir / path
# Security: prevent path traversal outside working dir
resolved = file_path.resolve()
if not str(resolved).startswith(str(self.working_dir.resolve())):
return ToolResult(
success=False,
error="Access denied: path outside working directory",
)
if not resolved.exists():
return ToolResult(
success=False,
error=f"File not found: {path}",
)
if not resolved.is_file():
return ToolResult(
success=False,
error=f"Not a file: {path}",
)
size = resolved.stat().st_size
if size > self.max_file_size:
return ToolResult(
success=False,
error=f"File too large: {size} bytes (max {self.max_file_size})",
)
content = resolved.read_text(encoding="utf-8", errors="replace")
return ToolResult(
success=True,
output=content,
metadata={"path": str(resolved), "size": size},
)
except Exception as e:
return ToolResult(
success=False,
error=f"Failed to read file: {str(e)}",
)
class WriteFileTool(Tool):
"""Write content to a file."""
def __init__(
self,
working_dir: Optional[str] = None,
max_file_size: int = 100000,
):
self.working_dir = Path(working_dir) if working_dir else Path.cwd()
self.max_file_size = max_file_size
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="write_file",
description="Write content to a file at the given path. Creates parent directories if needed.",
parameters={
"path": {
"type": "string",
"description": "Path to the file (relative to working directory)",
},
"content": {
"type": "string",
"description": "Content to write to the file",
},
},
required=["path", "content"],
)
async def execute(self, path: str, content: str) -> ToolResult:
"""Write content to a file."""
try:
if len(content) > self.max_file_size:
return ToolResult(
success=False,
error=f"Content too large: {len(content)} bytes (max {self.max_file_size})",
)
file_path = self.working_dir / path
# Security: prevent path traversal outside working dir
resolved = file_path.resolve()
if not str(resolved).startswith(str(self.working_dir.resolve())):
return ToolResult(
success=False,
error="Access denied: path outside working directory",
)
# Create parent directories
resolved.parent.mkdir(parents=True, exist_ok=True)
resolved.write_text(content, encoding="utf-8")
return ToolResult(
success=True,
output=f"Successfully wrote {len(content)} bytes to {path}",
metadata={"path": str(resolved), "size": len(content)},
)
except Exception as e:
return ToolResult(
success=False,
error=f"Failed to write file: {str(e)}",
)
class FireCrawl(Tool):
"""Perform a web crawl using FireCrawl tool."""
def __init__(
self,
working_dir: Optional[str] = None,
timeout: float = 60.0,
):
self.working_dir = working_dir or os.getcwd()
self.timeout = timeout
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="firecrawl",
description="Perform a web crawl starting from a given URL using FireCrawl.",
parameters={
"start_url": {
"type": "string",
"description": "The starting URL for the web crawl",
},
"max_pages": {
"type": "integer",
"description": "Maximum number of pages to crawl",
},
},
required=["start_url"],
)
async def execute(self, start_url: str, max_pages: int = 100) -> ToolResult:
"""Execute a web crawl using FireCrawl."""
try:
command = f"firecrawl --start-url {start_url} --max-pages {max_pages}"
bash_tool = BashTool(working_dir=self.working_dir, timeout=self.timeout)
result = await bash_tool.execute(command)
return result
except Exception as e:
return ToolResult(
success=False,
error=f"Failed to execute FireCrawl: {str(e)}",
)

64
atropos/tools/build_registry.py Normal file
View File

@ -0,0 +1,64 @@
"""
Unified tool registry builder for Hermes-Agent Atropos integration.
This composes:
- sandbox tool stubs (terminal/bash/read_file/write_file + stateful terminal/tmux)
- Hermes external tools (web/vision/image/moa/skills/browser), executed via ToolServer
ToolExecutor only needs the schema + `external` routing bit; ToolServer executes
the external tools via Hermes' existing implementations.
"""
from __future__ import annotations
from typing import List, Optional
from .base import ToolRegistry
from .hermes_external_tools import build_external_tools
from .sandbox_stubs import BashTool, ReadFileTool, TerminalTool, WriteFileTool
from .terminal_stateful_tool import TerminalStatefulTool
from .tmux_tool import TmuxTool
from .toolset_resolver import resolve_multiple_toolsets
def build_tool_registry(
*,
enabled_toolsets: Optional[List[str]] = None,
disabled_toolsets: Optional[List[str]] = None,
tool_server_url: Optional[str] = None,
) -> ToolRegistry:
"""
Build a ToolRegistry for AgentEnv / ToolExecutor / ToolServer.
If `tool_server_url` is not provided, external tools will be omitted so we do
not advertise tools that cannot execute.
"""
enabled_toolsets = enabled_toolsets or ["default"]
# Resolve tool names using Hermes toolsets plus Atropos additions.
selected = set(resolve_multiple_toolsets(enabled_toolsets))
if disabled_toolsets:
selected -= set(resolve_multiple_toolsets(disabled_toolsets))
reg = ToolRegistry()
# Always register sandbox tools if selected.
sandbox_by_name = {
"terminal": TerminalTool(),
"bash": BashTool(),
"read_file": ReadFileTool(),
"write_file": WriteFileTool(),
"terminal_stateful": TerminalStatefulTool(),
"tmux": TmuxTool(),
}
for name, tool in sandbox_by_name.items():
if name in selected:
reg.register(tool)
# External tools: only include when ToolServer is configured.
if tool_server_url:
for tool in build_external_tools(selected_tool_names=selected):
if tool.name in selected:
reg.register(tool)
return reg

90
atropos/tools/hermes_external_tools.py Normal file
View File

@ -0,0 +1,90 @@
"""
Hermes external tool adapter for Atropos ToolServer.
These tools reuse Hermes-Agent's existing tool runner (`model_tools.handle_function_call`)
so we don't duplicate external tool implementations.
Important:
- These are marked `external=True` and should be executed ONLY by ToolServer.
- We run `handle_function_call` in a worker thread because the Hermes implementation
uses `asyncio.run()` internally for some async tools (web_extract, vision, MoA, etc).
"""
from __future__ import annotations
import asyncio
import json
from typing import Any, Dict, List, Optional
import model_tools
from .base import Tool, ToolResult, ToolSchema
def _schema_from_openai_tool_dict(tool: Dict[str, Any], *, external: bool) -> ToolSchema:
fn = tool.get("function") or {}
name = str(fn.get("name") or "")
description = str(fn.get("description") or "")
params = fn.get("parameters") or {}
properties = params.get("properties") or {}
required = params.get("required") or []
if not isinstance(required, list):
required = []
return ToolSchema(
name=name,
description=description,
parameters=dict(properties),
required=[str(x) for x in required if isinstance(x, (str, int))],
external=external,
)
class HermesExternalTool(Tool):
def __init__(self, schema: ToolSchema):
self._schema = schema
@property
def schema(self) -> ToolSchema:
return self._schema
async def execute(self, task_id: Optional[str] = None, **kwargs: Any) -> ToolResult:
# `model_tools.handle_function_call` returns a JSON string (success or error).
# Run in a thread because some Hermes tool handlers call `asyncio.run()`.
raw = await asyncio.to_thread(model_tools.handle_function_call, self.name, kwargs, task_id)
try:
parsed = json.loads(raw)
except Exception:
# Keep as plain string.
return ToolResult(success=True, output=str(raw))
if isinstance(parsed, dict) and parsed.get("error"):
return ToolResult(success=False, error=str(parsed.get("error")), output="")
return ToolResult(success=True, output=json.dumps(parsed, ensure_ascii=False))
def build_external_tools(
*,
selected_tool_names: Optional[set[str]] = None,
) -> List[HermesExternalTool]:
"""
Build external tool wrappers from Hermes tool declarations.
Filters out sandbox-oriented tools (e.g. `terminal`) since those should run
inside the sandbox via ToolExecutor.
"""
# IMPORTANT: Hermes' `model_tools.get_tool_definitions()` only understands Hermes toolsets.
# Atropos envs add extra toolsets (filesystem/sandbox/stateful). To avoid noisy "Unknown toolset"
# prints and accidental filtering, we fetch ALL Hermes tool definitions here and filter by name.
tools = model_tools.get_tool_definitions(enabled_toolsets=None, disabled_toolsets=None, quiet_mode=True)
wrappers: List[HermesExternalTool] = []
for t in tools:
schema = _schema_from_openai_tool_dict(t, external=True)
if schema.name in {"terminal"}:
continue
if selected_tool_names is not None and schema.name not in selected_tool_names:
continue
wrappers.append(HermesExternalTool(schema))
return wrappers

129
atropos/tools/image_generation_tool.py
View File

@ -1,129 +0,0 @@
"""
Image generation tool (external).
This is intentionally minimal for the Phase 4.6 "external tool demo":
- executed via ToolServer (no secrets in sandboxes)
- by default returns a tiny inline PNG data URL (no network required)
- can optionally proxy to an OpenAI-compatible images endpoint (e.g. a local service)
"""
from __future__ import annotations
import base64
import json
import os
from dataclasses import dataclass
from typing import Literal, Optional
import httpx
from .base import Tool, ToolResult, ToolSchema
def _tiny_png_data_url() -> str:
# 1x1 transparent PNG
png_bytes = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMB/6X0kQAAAABJRU5ErkJggg=="
)
return f"data:image/png;base64,{base64.b64encode(png_bytes).decode('ascii')}"
@dataclass
class ImageGenerateConfig:
backend: Literal["inline", "openai"] = "inline"
base_url: Optional[str] = None
model: Optional[str] = None
timeout_s: float = 120.0
@classmethod
def from_env(cls) -> "ImageGenerateConfig":
backend = (os.getenv("IMAGE_GENERATE_BACKEND") or "inline").strip().lower()
if backend not in {"inline", "openai"}:
backend = "inline"
base_url = os.getenv("IMAGE_GENERATE_BASE_URL") or os.getenv("OLLAMA_BASE_URL")
model = os.getenv("IMAGE_GENERATE_MODEL") or os.getenv("OLLAMA_IMAGE_MODEL")
timeout_s = float(os.getenv("IMAGE_GENERATE_TIMEOUT_S", "120.0"))
return cls(
backend=backend, # type: ignore[arg-type]
base_url=base_url,
model=model,
timeout_s=timeout_s,
)
class ImageGenerateTool(Tool):
def __init__(self, config: Optional[ImageGenerateConfig] = None) -> None:
self._config = config or ImageGenerateConfig.from_env()
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="image_generate",
description=(
"Generate an image from a text prompt. Returns a JSON object containing an image URL "
"(often a data URL) plus metadata."
),
parameters={
"prompt": {"type": "string", "description": "The image prompt. Be detailed."},
"aspect_ratio": {
"type": "string",
"enum": ["landscape", "square", "portrait"],
"description": "Desired aspect ratio.",
"default": "landscape",
},
},
required=["prompt"],
external=True,
)
def is_available(self) -> tuple[bool, str | None]:
# Availability of this external tool is primarily governed by whether a ToolServer
# is configured/enabled. The tool itself will return a clear error if an upstream
# image endpoint is required but not configured.
return True, None
async def execute(self, prompt: str, aspect_ratio: str = "landscape") -> ToolResult: # noqa: ARG002
cfg = self._config
if cfg.backend == "inline":
payload = {
"url": _tiny_png_data_url(),
"backend": "inline",
"width": 1,
"height": 1,
}
return ToolResult(success=True, output=json.dumps(payload))
# OpenAI-compatible images generation endpoint.
# Expected: POST {base_url}/v1/images/generations with OpenAI-style body.
base_url = cfg.base_url or ""
base_url = base_url.rstrip("/")
if not base_url.endswith("/v1"):
base_url = f"{base_url}/v1"
url = f"{base_url}/images/generations"
body = {
"prompt": prompt,
"n": 1,
"response_format": "url",
}
if cfg.model:
body["model"] = cfg.model
try:
async with httpx.AsyncClient(timeout=cfg.timeout_s) as client:
resp = await client.post(url, json=body)
resp.raise_for_status()
data = resp.json()
except Exception as e:
return ToolResult(success=False, error=f"image_generate upstream call failed: {e}")
try:
image_url = data["data"][0]["url"]
except Exception:
return ToolResult(success=False, error=f"Unexpected image response format: {data!r}")
payload = {"url": image_url, "backend": "openai"}
return ToolResult(success=True, output=json.dumps(payload))

649
atropos/tools/mixture_of_agents_tool.py
View File

@ -1,649 +0,0 @@
#!/usr/bin/env python3
"""
Mixture-of-Agents Tool Module
This module implements the Mixture-of-Agents (MoA) methodology that leverages
the collective strengths of multiple LLMs through a layered architecture to
achieve state-of-the-art performance on complex reasoning tasks.
Based on the research paper: "Mixture-of-Agents Enhances Large Language Model Capabilities"
by Junlin Wang et al. (arXiv:2406.04692v1)
Key Features:
- Multi-layer LLM collaboration for enhanced reasoning
- Parallel processing of reference models for efficiency
- Intelligent aggregation and synthesis of diverse responses
- Specialized for extremely difficult problems requiring intense reasoning
- Optimized for coding, mathematics, and complex analytical tasks
Available Tool:
- mixture_of_agents_tool: Process complex queries using multiple frontier models
Architecture:
1. Reference models generate diverse initial responses in parallel
2. Aggregator model synthesizes responses into a high-quality output
3. Multiple layers can be used for iterative refinement (future enhancement)
Models Used (via OpenRouter):
- Reference Models: claude-opus-4, gemini-2.5-pro, gpt-4.1, deepseek-r1
- Aggregator Model: claude-opus-4 (highest capability for synthesis)
Configuration:
To customize the MoA setup, modify the configuration constants at the top of this file:
- REFERENCE_MODELS: List of models for generating diverse initial responses
- AGGREGATOR_MODEL: Model used to synthesize the final response
- REFERENCE_TEMPERATURE/AGGREGATOR_TEMPERATURE: Sampling temperatures
- MIN_SUCCESSFUL_REFERENCES: Minimum successful models needed to proceed
Usage:
from mixture_of_agents_tool import mixture_of_agents_tool
import asyncio
# Process a complex query
result = await mixture_of_agents_tool(
user_prompt="Solve this complex mathematical proof..."
)
"""
import json
import os
import asyncio
import uuid
import datetime
from pathlib import Path
from typing import Dict, Any, List, Optional
from openai import AsyncOpenAI
# Initialize OpenRouter API client for MoA processing
openrouter_client = AsyncOpenAI(
api_key=os.getenv("OPENROUTER_API_KEY"),
base_url="https://openrouter.ai/api/v1"
)
# Configuration for MoA processing
# Reference models - these generate diverse initial responses in parallel (OpenRouter slugs)
REFERENCE_MODELS = [
"anthropic/claude-opus-4.5",
"google/gemini-3-pro-preview",
"openai/gpt-5.2-pro",
"deepseek/deepseek-v3.2"
]
# Aggregator model - synthesizes reference responses into final output
AGGREGATOR_MODEL = "anthropic/claude-opus-4.5" # Use highest capability model for aggregation
# Temperature settings optimized for MoA performance
REFERENCE_TEMPERATURE = 0.6 # Balanced creativity for diverse perspectives
AGGREGATOR_TEMPERATURE = 0.4 # Focused synthesis for consistency
# Failure handling configuration
MIN_SUCCESSFUL_REFERENCES = 1 # Minimum successful reference models needed to proceed
# System prompt for the aggregator model (from the research paper)
AGGREGATOR_SYSTEM_PROMPT = """You have been provided with a set of responses from various open-source models to the latest user query. Your task is to synthesize these responses into a single, high-quality response. It is crucial to critically evaluate the information provided in these responses, recognizing that some of it may be biased or incorrect. Your response should not simply replicate the given answers but should offer a refined, accurate, and comprehensive reply to the instruction. Ensure your response is well-structured, coherent, and adheres to the highest standards of accuracy and reliability.
Responses from models:"""
# Debug mode configuration
DEBUG_MODE = os.getenv("MOA_TOOLS_DEBUG", "false").lower() == "true"
DEBUG_SESSION_ID = str(uuid.uuid4())
DEBUG_LOG_PATH = Path("./logs")
DEBUG_DATA = {
"session_id": DEBUG_SESSION_ID,
"start_time": datetime.datetime.now().isoformat(),
"debug_enabled": DEBUG_MODE,
"tool_calls": []
} if DEBUG_MODE else None
# Create logs directory if debug mode is enabled
if DEBUG_MODE:
DEBUG_LOG_PATH.mkdir(exist_ok=True)
print(f"🐛 MoA debug mode enabled - Session ID: {DEBUG_SESSION_ID}")
def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None:
"""
Log a debug call entry to the global debug data structure.
Args:
tool_name (str): Name of the tool being called
call_data (Dict[str, Any]): Data about the call including parameters and results
"""
if not DEBUG_MODE or not DEBUG_DATA:
return
call_entry = {
"timestamp": datetime.datetime.now().isoformat(),
"tool_name": tool_name,
**call_data
}
DEBUG_DATA["tool_calls"].append(call_entry)
def _save_debug_log() -> None:
"""
Save the current debug data to a JSON file in the logs directory.
"""
if not DEBUG_MODE or not DEBUG_DATA:
return
try:
debug_filename = f"moa_tools_debug_{DEBUG_SESSION_ID}.json"
debug_filepath = DEBUG_LOG_PATH / debug_filename
# Update end time
DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat()
DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"])
with open(debug_filepath, 'w', encoding='utf-8') as f:
json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False)
print(f"🐛 MoA debug log saved: {debug_filepath}")
except Exception as e:
print(f"❌ Error saving MoA debug log: {str(e)}")
def _construct_aggregator_prompt(system_prompt: str, responses: List[str]) -> str:
"""
Construct the final system prompt for the aggregator including all model responses.
Args:
system_prompt (str): Base system prompt for aggregation
responses (List[str]): List of responses from reference models
Returns:
str: Complete system prompt with enumerated responses
"""
response_text = "\n".join([f"{i+1}. {response}" for i, response in enumerate(responses)])
return f"{system_prompt}\n\n{response_text}"
async def _run_reference_model_safe(
model: str,
user_prompt: str,
temperature: float = REFERENCE_TEMPERATURE,
max_tokens: int = 32000,
max_retries: int = 6
) -> tuple[str, str, bool]:
"""
Run a single reference model with retry logic and graceful failure handling.
Args:
model (str): Model identifier to use
user_prompt (str): The user's query
temperature (float): Sampling temperature for response generation
max_tokens (int): Maximum tokens in response
max_retries (int): Maximum number of retry attempts
Returns:
tuple[str, str, bool]: (model_name, response_content_or_error, success_flag)
"""
for attempt in range(max_retries):
try:
print(f"🤖 Querying {model} (attempt {attempt + 1}/{max_retries})")
# Build parameters for the API call
api_params = {
"model": model,
"messages": [{"role": "user", "content": user_prompt}],
"extra_body": {
"reasoning": {
"enabled": True,
"effort": "xhigh"
}
}
}
# GPT models (especially gpt-4o-mini) don't support custom temperature values
# Only include temperature for non-GPT models
if not model.lower().startswith('gpt-'):
api_params["temperature"] = temperature
response = await openrouter_client.chat.completions.create(**api_params)
content = response.choices[0].message.content.strip()
print(f"✅ {model} responded ({len(content)} characters)")
return model, content, True
except Exception as e:
error_str = str(e)
# Log more detailed error information for debugging
if "invalid" in error_str.lower():
print(f"⚠️ {model} invalid request error (attempt {attempt + 1}): {error_str}")
elif "rate" in error_str.lower() or "limit" in error_str.lower():
print(f"⚠️ {model} rate limit error (attempt {attempt + 1}): {error_str}")
else:
print(f"⚠️ {model} unknown error (attempt {attempt + 1}): {error_str}")
if attempt < max_retries - 1:
# Exponential backoff for rate limiting: 2s, 4s, 8s, 16s, 32s, 60s
sleep_time = min(2 ** (attempt + 1), 60)
print(f" Retrying in {sleep_time}s...")
await asyncio.sleep(sleep_time)
else:
error_msg = f"{model} failed after {max_retries} attempts: {error_str}"
print(f"❌ {error_msg}")
return model, error_msg, False
async def _run_aggregator_model(
system_prompt: str,
user_prompt: str,
temperature: float = AGGREGATOR_TEMPERATURE,
max_tokens: int = None
) -> str:
"""
Run the aggregator model to synthesize the final response.
Args:
system_prompt (str): System prompt with all reference responses
user_prompt (str): Original user query
temperature (float): Focused temperature for consistent aggregation
max_tokens (int): Maximum tokens in final response
Returns:
str: Synthesized final response
"""
print(f"🧠 Running aggregator model: {AGGREGATOR_MODEL}")
# Build parameters for the API call
api_params = {
"model": AGGREGATOR_MODEL,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"extra_body": {
"reasoning": {
"enabled": True,
"effort": "xhigh"
}
}
}
# GPT models (especially gpt-4o-mini) don't support custom temperature values
# Only include temperature for non-GPT models
if not AGGREGATOR_MODEL.lower().startswith('gpt-'):
api_params["temperature"] = temperature
response = await openrouter_client.chat.completions.create(**api_params)
content = response.choices[0].message.content.strip()
print(f"✅ Aggregation complete ({len(content)} characters)")
return content
async def mixture_of_agents_tool(
user_prompt: str,
reference_models: Optional[List[str]] = None,
aggregator_model: Optional[str] = None
) -> str:
"""
Process a complex query using the Mixture-of-Agents methodology.
This tool leverages multiple frontier language models to collaboratively solve
extremely difficult problems requiring intense reasoning. It's particularly
effective for:
- Complex mathematical proofs and calculations
- Advanced coding problems and algorithm design
- Multi-step analytical reasoning tasks
- Problems requiring diverse domain expertise
- Tasks where single models show limitations
The MoA approach uses a fixed 2-layer architecture:
1. Layer 1: Multiple reference models generate diverse responses in parallel (temp=0.6)
2. Layer 2: Aggregator model synthesizes the best elements into final response (temp=0.4)
Args:
user_prompt (str): The complex query or problem to solve
reference_models (Optional[List[str]]): Custom reference models to use
aggregator_model (Optional[str]): Custom aggregator model to use
Returns:
str: JSON string containing the MoA results with the following structure:
{
"success": bool,
"response": str,
"models_used": {
"reference_models": List[str],
"aggregator_model": str
},
"processing_time": float
}
Raises:
Exception: If MoA processing fails or API key is not set
"""
start_time = datetime.datetime.now()
debug_call_data = {
"parameters": {
"user_prompt": user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt,
"reference_models": reference_models or REFERENCE_MODELS,
"aggregator_model": aggregator_model or AGGREGATOR_MODEL,
"reference_temperature": REFERENCE_TEMPERATURE,
"aggregator_temperature": AGGREGATOR_TEMPERATURE,
"min_successful_references": MIN_SUCCESSFUL_REFERENCES
},
"error": None,
"success": False,
"reference_responses_count": 0,
"failed_models_count": 0,
"failed_models": [],
"final_response_length": 0,
"processing_time_seconds": 0,
"models_used": {}
}
try:
print(f"🚀 Starting Mixture-of-Agents processing...")
print(f"📝 Query: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}")
# Validate API key availability
if not os.getenv("OPENROUTER_API_KEY"):
raise ValueError("OPENROUTER_API_KEY environment variable not set")
# Use provided models or defaults
ref_models = reference_models or REFERENCE_MODELS
agg_model = aggregator_model or AGGREGATOR_MODEL
print(f"🔄 Using {len(ref_models)} reference models in 2-layer MoA architecture")
# Layer 1: Generate diverse responses from reference models (with failure handling)
print("📡 Layer 1: Generating reference responses...")
model_results = await asyncio.gather(*[
_run_reference_model_safe(model, user_prompt, REFERENCE_TEMPERATURE)
for model in ref_models
])
# Separate successful and failed responses
successful_responses = []
failed_models = []
for model_name, content, success in model_results:
if success:
successful_responses.append(content)
else:
failed_models.append(model_name)
successful_count = len(successful_responses)
failed_count = len(failed_models)
print(f"📊 Reference model results: {successful_count} successful, {failed_count} failed")
if failed_models:
print(f"⚠️ Failed models: {', '.join(failed_models)}")
# Check if we have enough successful responses to proceed
if successful_count < MIN_SUCCESSFUL_REFERENCES:
raise ValueError(f"Insufficient successful reference models ({successful_count}/{len(ref_models)}). Need at least {MIN_SUCCESSFUL_REFERENCES} successful responses.")
debug_call_data["reference_responses_count"] = successful_count
debug_call_data["failed_models_count"] = failed_count
debug_call_data["failed_models"] = failed_models
# Layer 2: Aggregate responses using the aggregator model
print("🧠 Layer 2: Synthesizing final response...")
aggregator_system_prompt = _construct_aggregator_prompt(
AGGREGATOR_SYSTEM_PROMPT,
successful_responses
)
final_response = await _run_aggregator_model(
aggregator_system_prompt,
user_prompt,
AGGREGATOR_TEMPERATURE
)
# Calculate processing time
end_time = datetime.datetime.now()
processing_time = (end_time - start_time).total_seconds()
print(f"✅ MoA processing completed in {processing_time:.2f} seconds")
# Prepare successful response (only final aggregated result, minimal fields)
result = {
"success": True,
"response": final_response,
"models_used": {
"reference_models": ref_models,
"aggregator_model": agg_model
}
}
debug_call_data["success"] = True
debug_call_data["final_response_length"] = len(final_response)
debug_call_data["processing_time_seconds"] = processing_time
debug_call_data["models_used"] = result["models_used"]
# Log debug information
_log_debug_call("mixture_of_agents_tool", debug_call_data)
_save_debug_log()
return json.dumps(result, indent=2, ensure_ascii=False)
except Exception as e:
error_msg = f"Error in MoA processing: {str(e)}"
print(f"❌ {error_msg}")
# Calculate processing time even for errors
end_time = datetime.datetime.now()
processing_time = (end_time - start_time).total_seconds()
# Prepare error response (minimal fields)
result = {
"success": False,
"response": "MoA processing failed. Please try again or use a single model for this query.",
"models_used": {
"reference_models": reference_models or REFERENCE_MODELS,
"aggregator_model": aggregator_model or AGGREGATOR_MODEL
},
"error": error_msg
}
debug_call_data["error"] = error_msg
debug_call_data["processing_time_seconds"] = processing_time
_log_debug_call("mixture_of_agents_tool", debug_call_data)
_save_debug_log()
return json.dumps(result, indent=2, ensure_ascii=False)
def check_openrouter_api_key() -> bool:
"""
Check if the OpenRouter API key is available in environment variables.
Returns:
bool: True if API key is set, False otherwise
"""
return bool(os.getenv("OPENROUTER_API_KEY"))
def check_moa_requirements() -> bool:
"""
Check if all requirements for MoA tools are met.
Returns:
bool: True if requirements are met, False otherwise
"""
return check_openrouter_api_key()
def get_debug_session_info() -> Dict[str, Any]:
"""
Get information about the current debug session.
Returns:
Dict[str, Any]: Dictionary containing debug session information
"""
if not DEBUG_MODE or not DEBUG_DATA:
return {
"enabled": False,
"session_id": None,
"log_path": None,
"total_calls": 0
}
return {
"enabled": True,
"session_id": DEBUG_SESSION_ID,
"log_path": str(DEBUG_LOG_PATH / f"moa_tools_debug_{DEBUG_SESSION_ID}.json"),
"total_calls": len(DEBUG_DATA["tool_calls"])
}
def get_available_models() -> Dict[str, List[str]]:
"""
Get information about available models for MoA processing.
Returns:
Dict[str, List[str]]: Dictionary with reference and aggregator models
"""
return {
"reference_models": REFERENCE_MODELS,
"aggregator_models": [AGGREGATOR_MODEL],
"supported_models": REFERENCE_MODELS + [AGGREGATOR_MODEL]
}
def get_moa_configuration() -> Dict[str, Any]:
"""
Get the current MoA configuration settings.
Returns:
Dict[str, Any]: Dictionary containing all configuration parameters
"""
return {
"reference_models": REFERENCE_MODELS,
"aggregator_model": AGGREGATOR_MODEL,
"reference_temperature": REFERENCE_TEMPERATURE,
"aggregator_temperature": AGGREGATOR_TEMPERATURE,
"min_successful_references": MIN_SUCCESSFUL_REFERENCES,
"total_reference_models": len(REFERENCE_MODELS),
"failure_tolerance": f"{len(REFERENCE_MODELS) - MIN_SUCCESSFUL_REFERENCES}/{len(REFERENCE_MODELS)} models can fail"
}
# =============================================================================
# Atropos-Agent Tool wrapper (Hermes compatibility)
# =============================================================================
from .base import Tool, ToolResult, ToolSchema # noqa: E402
def _tool_result_from_json(output: str) -> ToolResult:
try:
data = json.loads(output)
except Exception:
return ToolResult(success=True, output=output)
if isinstance(data, dict) and "success" in data:
ok = bool(data.get("success"))
if not ok:
err = data.get("error") or "Mixture-of-Agents failed"
return ToolResult(success=False, output=output, error=str(err))
return ToolResult(success=True, output=output)
return ToolResult(success=True, output=output)
class MixtureOfAgentsTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="mixture_of_agents",
description=(
"Process extremely difficult problems requiring intense reasoning using a Mixture-of-Agents."
),
parameters={
"user_prompt": {
"type": "string",
"description": "The complex query or problem to solve using multiple models.",
}
},
required=["user_prompt"],
external=True,
)
def is_available(self) -> tuple[bool, str | None]:
if not os.getenv("OPENROUTER_API_KEY"):
return False, "OPENROUTER_API_KEY not set"
return True, None
async def execute(self, user_prompt: str) -> ToolResult:
output = await mixture_of_agents_tool(user_prompt=user_prompt)
return _tool_result_from_json(output)
if __name__ == "__main__":
"""
Simple test/demo when run directly
"""
print("🤖 Mixture-of-Agents Tool Module")
print("=" * 50)
# Check if API key is available
api_available = check_openrouter_api_key()
if not api_available:
print("❌ OPENROUTER_API_KEY environment variable not set")
print("Please set your API key: export OPENROUTER_API_KEY='your-key-here'")
print("Get API key at: https://openrouter.ai/")
exit(1)
else:
print("✅ OpenRouter API key found")
print("🛠️ MoA tools ready for use!")
# Show current configuration
config = get_moa_configuration()
print(f"\n⚙️ Current Configuration:")
print(f" 🤖 Reference models ({len(config['reference_models'])}): {', '.join(config['reference_models'])}")
print(f" 🧠 Aggregator model: {config['aggregator_model']}")
print(f" 🌡️ Reference temperature: {config['reference_temperature']}")
print(f" 🌡️ Aggregator temperature: {config['aggregator_temperature']}")
print(f" 🛡️ Failure tolerance: {config['failure_tolerance']}")
print(f" 📊 Minimum successful models: {config['min_successful_references']}")
# Show debug mode status
if DEBUG_MODE:
print(f"\n🐛 Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}")
print(f" Debug logs will be saved to: ./logs/moa_tools_debug_{DEBUG_SESSION_ID}.json")
else:
print("\n🐛 Debug mode disabled (set MOA_TOOLS_DEBUG=true to enable)")
print("\nBasic usage:")
print(" from mixture_of_agents_tool import mixture_of_agents_tool")
print(" import asyncio")
print("")
print(" async def main():")
print(" result = await mixture_of_agents_tool(")
print(" user_prompt='Solve this complex mathematical proof...'")
print(" )")
print(" print(result)")
print(" asyncio.run(main())")
print("\nBest use cases:")
print(" - Complex mathematical proofs and calculations")
print(" - Advanced coding problems and algorithm design")
print(" - Multi-step analytical reasoning tasks")
print(" - Problems requiring diverse domain expertise")
print(" - Tasks where single models show limitations")
print("\nPerformance characteristics:")
print(" - Higher latency due to multiple model calls")
print(" - Significantly improved quality for complex tasks")
print(" - Parallel processing for efficiency")
print(f" - Optimized temperatures: {REFERENCE_TEMPERATURE} for reference models, {AGGREGATOR_TEMPERATURE} for aggregation")
print(" - Token-efficient: only returns final aggregated response")
print(" - Resilient: continues with partial model failures")
print(f" - Configurable: easy to modify models and settings at top of file")
print(" - State-of-the-art results on challenging benchmarks")
print("\nDebug mode:")
print(" # Enable debug logging")
print(" export MOA_TOOLS_DEBUG=true")
print(" # Debug logs capture all MoA processing steps and metrics")
print(" # Logs saved to: ./logs/moa_tools_debug_UUID.json")

98
atropos/tools/sandbox_stubs.py Normal file
View File

@ -0,0 +1,98 @@
"""
Sandbox tool stubs for Atropos ToolExecutor.
These tools are executed inside the sandbox containers via:
ToolExecutor -> SlotPool -> sandbox_server.py
They intentionally do NOT execute anything on the host process. If they are
called directly (outside ToolExecutor), they return a clear error.
"""
from __future__ import annotations
from typing import Optional
from .base import Tool, ToolResult, ToolSchema
class TerminalTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="terminal",
description=(
"Execute a command inside the sandbox slot workspace and return stdout/stderr. "
"Filesystem persists within a trajectory slot. Background processes are not supported "
"in stateless mode."
),
parameters={
"command": {"type": "string", "description": "The command to execute"},
"timeout": {
"type": "integer",
"description": "Command timeout in seconds (optional).",
"minimum": 1,
},
"background": {
"type": "boolean",
"description": "Not supported in sandbox terminal (always false).",
"default": False,
},
},
required=["command"],
external=False,
)
async def execute(self, **_kwargs) -> ToolResult:
return ToolResult(
success=False,
error="terminal must be executed via ToolExecutor inside the sandbox",
)
class BashTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="bash",
description="Execute a bash command inside the sandbox slot workspace.",
parameters={"command": {"type": "string", "description": "The bash command to execute"}},
required=["command"],
external=False,
)
async def execute(self, **_kwargs) -> ToolResult:
return ToolResult(success=False, error="bash must be executed via ToolExecutor inside the sandbox")
class ReadFileTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="read_file",
description="Read a file from the sandbox slot workspace.",
parameters={"path": {"type": "string", "description": "Path to the file"}},
required=["path"],
external=False,
)
async def execute(self, **_kwargs) -> ToolResult:
return ToolResult(success=False, error="read_file must be executed via ToolExecutor inside the sandbox")
class WriteFileTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="write_file",
description="Write a file into the sandbox slot workspace.",
parameters={
"path": {"type": "string", "description": "Path to the file"},
"content": {"type": "string", "description": "File content"},
},
required=["path", "content"],
external=False,
)
async def execute(self, **_kwargs) -> ToolResult:
return ToolResult(success=False, error="write_file must be executed via ToolExecutor inside the sandbox")

437
atropos/tools/terminal_hecate.py
View File

@ -1,437 +0,0 @@
#!/usr/bin/env python3
"""
Terminal Hecate Tool Module
A terminal tool that executes commands on MorphCloud/Hecate VMs.
Uses E2B-style cloud VMs for execution with automatic lifecycle management.
Features:
- Direct SSH command execution on cloud VMs
- Background task support
- VM lifecycle management with TTL
- Automatic cleanup after inactivity
Usage:
from terminal_hecate import terminal_hecate_tool
# Execute a simple command
result = terminal_hecate_tool("ls -la")
# Execute in background
result = terminal_hecate_tool("python server.py", background=True)
"""
import json
import os
import time
import threading
import atexit
from typing import Optional, Dict, Any
# Tool description for LLM
TERMINAL_HECATE_DESCRIPTION = """Execute commands on a secure cloud Linux VM environment (Hecate/MorphCloud).
**Environment:**
- Minimal Debian-based OS with internet access
- Automatic VM lifecycle management (creates on-demand, reuses, cleans up)
- Filesystem is persisted between tool calls but environment variables, venvs, etc are reset.
**Command Execution:**
- Simple commands: Just provide the 'command' parameter
- Background processes: Set 'background': True for servers/long-running tasks
- Command timeout: Optional 'timeout' parameter in seconds
**Examples:**
- Run command: `{"command": "ls -la"}`
- Background task: `{"command": "source path/to/my/venv/bin/activate && python server.py", "background": True}`
- With timeout: `{"command": "long_task.sh", "timeout": 300}`
**Best Practices:**
- Run servers/long processes in background
- Monitor disk usage for large tasks
- Install whatever tools you need with sudo apt-get
- Do not be afraid to run pip with --break-system-packages
**Things to avoid**
- Do NOT use interactive tools such as tmux, vim, nano, python repl - you will get stuck. Even git sometimes becomes interactive if the output is large. If you're not sure pipe to cat.
"""
# Global state for VM lifecycle management
_active_instances: Dict[str, Any] = {}
_last_activity: Dict[str, float] = {}
_instance_lock = threading.Lock()
_cleanup_thread = None
_cleanup_running = False
def _cleanup_inactive_vms(vm_lifetime_seconds: int = 300):
"""Clean up VMs that have been inactive for longer than vm_lifetime_seconds."""
global _active_instances, _last_activity
current_time = time.time()
tasks_to_cleanup = []
with _instance_lock:
for task_id, last_time in list(_last_activity.items()):
if current_time - last_time > vm_lifetime_seconds:
tasks_to_cleanup.append(task_id)
for task_id in tasks_to_cleanup:
try:
if task_id in _active_instances:
instance = _active_instances[task_id]
if hasattr(instance, 'terminate'):
instance.terminate()
elif hasattr(instance, 'stop'):
instance.stop()
elif hasattr(instance, 'delete'):
instance.delete()
del _active_instances[task_id]
print(f"[VM Cleanup] Terminated inactive VM for task: {task_id}")
if task_id in _last_activity:
del _last_activity[task_id]
except Exception as e:
# 404 errors are benign - VM already cleaned up by TTL
error_str = str(e)
if "404" in error_str or "InstanceNotFoundError" in error_str or "not found" in error_str.lower():
print(f"[VM Cleanup] VM for task {task_id} already cleaned up (likely TTL expiration)")
else:
print(f"[VM Cleanup] Error cleaning up VM for task {task_id}: {e}")
# Always remove from tracking dicts to prevent infinite retry loops
if task_id in _active_instances:
del _active_instances[task_id]
if task_id in _last_activity:
del _last_activity[task_id]
def _cleanup_thread_worker():
"""Background thread worker that periodically cleans up inactive VMs."""
global _cleanup_running
while _cleanup_running:
try:
vm_lifetime = int(os.getenv("HECATE_VM_LIFETIME_SECONDS", "300"))
_cleanup_inactive_vms(vm_lifetime)
except Exception as e:
print(f"[VM Cleanup] Error in cleanup thread: {e}")
for _ in range(60):
if not _cleanup_running:
break
time.sleep(1)
def _start_cleanup_thread():
"""Start the background cleanup thread if not already running."""
global _cleanup_thread, _cleanup_running
with _instance_lock:
if _cleanup_thread is None or not _cleanup_thread.is_alive():
_cleanup_running = True
_cleanup_thread = threading.Thread(target=_cleanup_thread_worker, daemon=True)
_cleanup_thread.start()
def _stop_cleanup_thread():
"""Stop the background cleanup thread."""
global _cleanup_running
_cleanup_running = False
if _cleanup_thread is not None:
_cleanup_thread.join(timeout=5)
def cleanup_vm(task_id: str):
"""Manually clean up a specific VM by task_id."""
global _active_instances, _last_activity
with _instance_lock:
try:
if task_id in _active_instances:
instance = _active_instances[task_id]
if hasattr(instance, 'terminate'):
instance.terminate()
elif hasattr(instance, 'stop'):
instance.stop()
elif hasattr(instance, 'delete'):
instance.delete()
del _active_instances[task_id]
print(f"[VM Cleanup] Manually terminated VM for task: {task_id}")
if task_id in _last_activity:
del _last_activity[task_id]
except Exception as e:
# 404 errors are benign - VM already cleaned up by TTL
error_str = str(e)
if "404" in error_str or "InstanceNotFoundError" in error_str or "not found" in error_str.lower():
print(f"[VM Cleanup] VM for task {task_id} already cleaned up (likely TTL expiration)")
else:
print(f"[VM Cleanup] Error manually cleaning up VM for task {task_id}: {e}")
atexit.register(_stop_cleanup_thread)
def _execute_command(instance, command: str, timeout: Optional[int] = None) -> Dict[str, Any]:
"""
Execute a command on the VM instance using instance.exec() for proper stderr capture.
Args:
instance: MorphVM instance
command: Command to execute
timeout: Optional timeout in seconds (Note: exec() may not support timeout directly)
Returns:
dict with stdout, stderr, returncode
"""
try:
# Use instance.exec() which properly captures both stdout and stderr
# (unlike ssh.run() which doesn't capture stderr correctly)
result = instance.exec(command)
# Debug logging only for verbose mode or unusual cases
# Note: Non-zero exit codes are normal (model's command failed) - not a tool error
if result.exit_code != 0 and not result.stdout and not result.stderr:
# Only log if we got absolutely no output - might indicate an issue
print(f"⚠️ Command returned exit={result.exit_code} with no output")
return {
"stdout": result.stdout or "",
"stderr": result.stderr or "",
"returncode": result.exit_code
}
except Exception as e:
# Check if it's a timeout
error_str = str(e).lower()
if "timeout" in error_str:
return {
"stdout": "",
"stderr": f"Command timed out after {timeout or 120} seconds",
"returncode": 124
}
return {
"stdout": "",
"stderr": f"Command execution failed: {str(e)}",
"returncode": -1
}
def terminal_hecate_tool(
command: str,
background: bool = False,
timeout: Optional[int] = None,
task_id: Optional[str] = None
) -> str:
"""
Execute a command on a MorphCloud/Hecate VM without session persistence.
Args:
command: The command to execute
background: Whether to run in background (default: False)
timeout: Command timeout in seconds (default: 120)
task_id: Unique identifier for VM isolation (optional)
Returns:
str: JSON string with output, exit_code, and error fields
Examples:
# Execute a simple command
>>> result = terminal_hecate_tool(command="ls -la /tmp")
# Run a background task
>>> result = terminal_hecate_tool(command="python server.py", background=True)
# With custom timeout
>>> result = terminal_hecate_tool(command="long_task.sh", timeout=300)
"""
global _active_instances, _last_activity
try:
# Import required modules
try:
from morphcloud.api import MorphCloudClient
except ImportError as import_error:
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Terminal tool disabled: {import_error}",
"status": "disabled"
}, ensure_ascii=False)
# Get configuration
vm_ttl_seconds = int(os.getenv("HECATE_VM_TTL_SECONDS", "1200"))
snapshot_id = os.getenv("HECATE_DEFAULT_SNAPSHOT_ID", "snapshot_defv9tjg")
# Check API key
morph_api_key = os.getenv("MORPH_API_KEY")
if not morph_api_key:
return json.dumps({
"output": "",
"exit_code": -1,
"error": "MORPH_API_KEY environment variable not set",
"status": "disabled"
}, ensure_ascii=False)
# Use task_id for VM isolation
effective_task_id = task_id or "default"
# Start cleanup thread
_start_cleanup_thread()
# Get or create VM instance
with _instance_lock:
if effective_task_id not in _active_instances:
morph_client = MorphCloudClient(api_key=morph_api_key)
_active_instances[effective_task_id] = morph_client.instances.start(
snapshot_id=snapshot_id,
ttl_seconds=vm_ttl_seconds,
ttl_action="stop"
)
# Update last activity time
_last_activity[effective_task_id] = time.time()
instance = _active_instances[effective_task_id]
# Wait for instance to be ready
instance.wait_until_ready()
# Prepare command for execution
if background:
# Run in background with nohup and redirect output
exec_command = f"nohup {command} > /tmp/bg_output.log 2>&1 &"
result = _execute_command(instance, exec_command, timeout=10)
# For background tasks, return immediately with info
if result["returncode"] == 0:
return json.dumps({
"output": "Background task started successfully",
"exit_code": 0,
"error": None
}, ensure_ascii=False)
else:
# Include stderr in output but don't set error (command failure, not tool failure)
bg_output = result["stdout"]
if result["stderr"]:
bg_output = f"{bg_output}\n{result['stderr']}" if bg_output else result["stderr"]
return json.dumps({
"output": bg_output,
"exit_code": result["returncode"],
"error": None # Only set for actual tool failures
}, ensure_ascii=False)
else:
# Run foreground command with retry logic for transient failures
max_retries = 3
retry_count = 0
result = None
while retry_count <= max_retries:
result = _execute_command(instance, command, timeout=timeout)
# Check if we should retry (only for transient errors, not normal results)
stdout = result.get("stdout", "")
stderr = result.get("stderr", "")
returncode = result.get("returncode", 0)
should_retry = False
retry_reason = ""
# NOTE: Empty output with exit_code=0 is NORMAL for many commands:
# - File writes: cat > file, echo > file
# - Directory ops: mkdir, cd
# - Silent installs: pip install --quiet
# So we do NOT retry on exit_code=0, even with empty output.
# Only retry on special error codes that suggest transient/infra issues
if not stdout and not stderr and returncode in [-1, 124]:
should_retry = True
retry_reason = f"transient error (code {returncode})"
if should_retry and retry_count < max_retries:
retry_count += 1
wait_time = 2 ** retry_count # Exponential backoff: 2s, 4s, 8s
print(f"⚠️ Terminal: {retry_reason}, retrying in {wait_time}s (attempt {retry_count}/{max_retries})")
time.sleep(wait_time)
continue
# Got a result (success or normal command failure) - exit retry loop
break
# Combine stdout and stderr for output
output = result["stdout"]
if result["stderr"] and result["returncode"] != 0:
output = f"{output}\n{result['stderr']}" if output else result["stderr"]
# Truncate output if too long (max 50,000 chars to avoid context explosion)
MAX_OUTPUT_CHARS = 50000
if len(output) > MAX_OUTPUT_CHARS:
truncated_notice = f"\n\n... [OUTPUT TRUNCATED - showing last {MAX_OUTPUT_CHARS} chars of {len(output)} total] ..."
output = truncated_notice + output[-MAX_OUTPUT_CHARS:]
# NOTE: error is only set for FUNCTIONAL tool failures (VM issues, timeouts, etc.)
# Non-zero exit codes from the model's commands are NOT tool failures -
# the model can self-correct. The exit_code field tells the model if the command succeeded.
# Retries that eventually succeed also don't count as failures.
return json.dumps({
"output": output.strip(),
"exit_code": result["returncode"],
"error": None # Only set for actual tool failures, not command failures
}, ensure_ascii=False)
except Exception as e:
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Failed to execute command: {str(e)}",
"status": "error"
}, ensure_ascii=False)
def check_hecate_requirements() -> bool:
"""Check if all requirements for the Hecate terminal tool are met."""
required_vars = ["MORPH_API_KEY"]
missing_required = [var for var in required_vars if not os.getenv(var)]
if missing_required:
print(f"Missing required environment variables: {', '.join(missing_required)}")
return False
try:
from morphcloud.api import MorphCloudClient
return True
except Exception as e:
print(f"MorphCloud not available: {e}")
return False
if __name__ == "__main__":
"""Simple test when run directly."""
print("Terminal Hecate Tool Module (MorphCloud/E2B)")
print("=" * 40)
if not check_hecate_requirements():
print("Requirements not met. Please check the messages above.")
exit(1)
print("All requirements met!")
print("\nAvailable Tool:")
print(" - terminal_hecate_tool: Execute commands on cloud VMs")
print("\nUsage Examples:")
print(" # Execute a command")
print(" result = terminal_hecate_tool(command='ls -la')")
print(" ")
print(" # Run a background task")
print(" result = terminal_hecate_tool(command='python server.py', background=True)")
print("\nEnvironment Variables:")
print(f" MORPH_API_KEY: {'Set' if os.getenv('MORPH_API_KEY') else 'Not set'}")
print(f" HECATE_VM_TTL_SECONDS: {os.getenv('HECATE_VM_TTL_SECONDS', '1200')} (default: 1200 / 20 minutes)")
print(f" HECATE_VM_LIFETIME_SECONDS: {os.getenv('HECATE_VM_LIFETIME_SECONDS', '300')} (default: 300 / 5 minutes)")
print(f" HECATE_DEFAULT_SNAPSHOT_ID: {os.getenv('HECATE_DEFAULT_SNAPSHOT_ID', 'snapshot_defv9tjg')}")

10
atropos/tools/terminal_stateful_tool.py
View File

@ -11,7 +11,6 @@ from __future__ import annotations
from typing import Optional
from .base import Tool, ToolResult, ToolSchema
from .basic_tools import BashTool
class TerminalStatefulTool(Tool):
@ -39,7 +38,8 @@ class TerminalStatefulTool(Tool):
return True, None
async def execute(self, command: str, timeout: Optional[int] = None) -> ToolResult:
# Fallback direct execution (not stateful) when used outside ToolExecutor.
bash = BashTool(timeout=float(timeout) if timeout else 30.0)
return await bash.execute(command=command)
_ = (command, timeout)
return ToolResult(
success=False,
error="terminal_stateful must be executed via ToolExecutor inside the sandbox",
)

493
atropos/tools/terminal_tool.py
View File

@ -1,493 +0,0 @@
#!/usr/bin/env python3
"""
Terminal Tool Module (mini-swe-agent backend)
A terminal tool that executes commands using mini-swe-agent's execution environments.
Supports local execution, Docker containers, and Modal cloud sandboxes.
Environment Selection (via TERMINAL_ENV environment variable):
- "local": Execute directly on the host machine (default, fastest)
- "docker": Execute in Docker containers (isolated, requires Docker)
- "modal": Execute in Modal cloud sandboxes (scalable, requires Modal account)
Features:
- Multiple execution backends (local, docker, modal)
- Background task support
- VM/container lifecycle management
- Automatic cleanup after inactivity
Usage:
from terminal_tool import terminal_tool
# Execute a simple command
result = terminal_tool("ls -la")
# Execute in background
result = terminal_tool("python server.py", background=True)
"""
import json
import os
import sys
import time
import threading
import atexit
from pathlib import Path
from typing import Optional, Dict, Any
# Add mini-swe-agent to path if not installed
mini_swe_path = Path(__file__).parent.parent / "mini-swe-agent" / "src"
if mini_swe_path.exists():
sys.path.insert(0, str(mini_swe_path))
# Tool description for LLM
TERMINAL_TOOL_DESCRIPTION = """Execute commands on a secure Linux environment.
**Environment:**
- Isolated execution environment (local, Docker, or Modal cloud based on configuration)
- Filesystem persists between tool calls within the same task
- Internet access available
**Command Execution:**
- Simple commands: Just provide the 'command' parameter
- Background processes: Set 'background': True for servers/long-running tasks
- Command timeout: Optional 'timeout' parameter in seconds
**Examples:**
- Run command: `{"command": "ls -la"}`
- Background task: `{"command": "source venv/bin/activate && python server.py", "background": True}`
- With timeout: `{"command": "long_task.sh", "timeout": 300}`
**Best Practices:**
- Run servers/long processes in background
- Monitor disk usage for large tasks
- Install whatever tools you need with apt-get or pip
- Do not be afraid to run pip with --break-system-packages
**Things to avoid:**
- Do NOT use interactive tools such as tmux, vim, nano, python repl - you will get stuck.
- Even git sometimes becomes interactive if the output is large. If you're not sure, pipe to cat.
"""
# Global state for environment lifecycle management
_active_environments: Dict[str, Any] = {}
_last_activity: Dict[str, float] = {}
_env_lock = threading.Lock()
_cleanup_thread = None
_cleanup_running = False
# Configuration from environment variables
def _get_env_config() -> Dict[str, Any]:
"""Get terminal environment configuration from environment variables."""
return {
"env_type": os.getenv("TERMINAL_ENV", "local"), # local, docker, or modal
"docker_image": os.getenv("TERMINAL_DOCKER_IMAGE", "python:3.11-slim"),
"modal_image": os.getenv("TERMINAL_MODAL_IMAGE", "python:3.11-slim"),
"cwd": os.getenv("TERMINAL_CWD", "/tmp"),
"timeout": int(os.getenv("TERMINAL_TIMEOUT", "60")),
"lifetime_seconds": int(os.getenv("TERMINAL_LIFETIME_SECONDS", "300")),
}
def _create_environment(env_type: str, image: str, cwd: str, timeout: int):
"""
Create an execution environment from mini-swe-agent.
Args:
env_type: One of "local", "docker", "modal"
image: Docker/Modal image name (ignored for local)
cwd: Working directory
timeout: Default command timeout
Returns:
Environment instance with execute() method
"""
if env_type == "local":
from minisweagent.environments.local import LocalEnvironment
return LocalEnvironment(cwd=cwd, timeout=timeout)
elif env_type == "docker":
from minisweagent.environments.docker import DockerEnvironment
return DockerEnvironment(image=image, cwd=cwd, timeout=timeout)
elif env_type == "modal":
from minisweagent.environments.extra.swerex_modal import SwerexModalEnvironment
return SwerexModalEnvironment(image=image, cwd=cwd, timeout=timeout)
else:
raise ValueError(f"Unknown environment type: {env_type}. Use 'local', 'docker', or 'modal'")
def _cleanup_inactive_envs(lifetime_seconds: int = 300):
"""Clean up environments that have been inactive for longer than lifetime_seconds."""
global _active_environments, _last_activity
current_time = time.time()
tasks_to_cleanup = []
with _env_lock:
for task_id, last_time in list(_last_activity.items()):
if current_time - last_time > lifetime_seconds:
tasks_to_cleanup.append(task_id)
for task_id in tasks_to_cleanup:
try:
if task_id in _active_environments:
env = _active_environments[task_id]
# Try various cleanup methods
if hasattr(env, 'cleanup'):
env.cleanup()
elif hasattr(env, 'stop'):
env.stop()
elif hasattr(env, 'terminate'):
env.terminate()
del _active_environments[task_id]
print(f"[Terminal Cleanup] Cleaned up inactive environment for task: {task_id}")
if task_id in _last_activity:
del _last_activity[task_id]
except Exception as e:
error_str = str(e)
if "404" in error_str or "not found" in error_str.lower():
print(f"[Terminal Cleanup] Environment for task {task_id} already cleaned up")
else:
print(f"[Terminal Cleanup] Error cleaning up environment for task {task_id}: {e}")
# Always remove from tracking dicts
if task_id in _active_environments:
del _active_environments[task_id]
if task_id in _last_activity:
del _last_activity[task_id]
def _cleanup_thread_worker():
"""Background thread worker that periodically cleans up inactive environments."""
global _cleanup_running
while _cleanup_running:
try:
config = _get_env_config()
_cleanup_inactive_envs(config["lifetime_seconds"])
except Exception as e:
print(f"[Terminal Cleanup] Error in cleanup thread: {e}")
for _ in range(60):
if not _cleanup_running:
break
time.sleep(1)
def _start_cleanup_thread():
"""Start the background cleanup thread if not already running."""
global _cleanup_thread, _cleanup_running
with _env_lock:
if _cleanup_thread is None or not _cleanup_thread.is_alive():
_cleanup_running = True
_cleanup_thread = threading.Thread(target=_cleanup_thread_worker, daemon=True)
_cleanup_thread.start()
def _stop_cleanup_thread():
"""Stop the background cleanup thread."""
global _cleanup_running
_cleanup_running = False
if _cleanup_thread is not None:
_cleanup_thread.join(timeout=5)
def cleanup_vm(task_id: str):
"""Manually clean up a specific environment by task_id."""
global _active_environments, _last_activity
with _env_lock:
try:
if task_id in _active_environments:
env = _active_environments[task_id]
if hasattr(env, 'cleanup'):
env.cleanup()
elif hasattr(env, 'stop'):
env.stop()
elif hasattr(env, 'terminate'):
env.terminate()
del _active_environments[task_id]
print(f"[Terminal Cleanup] Manually cleaned up environment for task: {task_id}")
if task_id in _last_activity:
del _last_activity[task_id]
except Exception as e:
error_str = str(e)
if "404" in error_str or "not found" in error_str.lower():
print(f"[Terminal Cleanup] Environment for task {task_id} already cleaned up")
else:
print(f"[Terminal Cleanup] Error cleaning up environment for task {task_id}: {e}")
atexit.register(_stop_cleanup_thread)
def terminal_tool(
command: str,
background: bool = False,
timeout: Optional[int] = None,
task_id: Optional[str] = None
) -> str:
"""
Execute a command using mini-swe-agent's execution environments.
Args:
command: The command to execute
background: Whether to run in background (default: False)
timeout: Command timeout in seconds (default: from config)
task_id: Unique identifier for environment isolation (optional)
Returns:
str: JSON string with output, exit_code, and error fields
Examples:
# Execute a simple command
>>> result = terminal_tool(command="ls -la /tmp")
# Run a background task
>>> result = terminal_tool(command="python server.py", background=True)
# With custom timeout
>>> result = terminal_tool(command="long_task.sh", timeout=300)
"""
global _active_environments, _last_activity
try:
# Get configuration
config = _get_env_config()
env_type = config["env_type"]
# Select image based on env type
if env_type == "docker":
image = config["docker_image"]
elif env_type == "modal":
image = config["modal_image"]
else:
image = ""
cwd = config["cwd"]
default_timeout = config["timeout"]
effective_timeout = timeout or default_timeout
# Use task_id for environment isolation
effective_task_id = task_id or "default"
# Start cleanup thread
_start_cleanup_thread()
# Get or create environment
with _env_lock:
if effective_task_id not in _active_environments:
try:
_active_environments[effective_task_id] = _create_environment(
env_type=env_type,
image=image,
cwd=cwd,
timeout=effective_timeout
)
except ImportError as e:
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Terminal tool disabled: mini-swe-agent not available ({e})",
"status": "disabled"
}, ensure_ascii=False)
# Update last activity time
_last_activity[effective_task_id] = time.time()
env = _active_environments[effective_task_id]
# Prepare command for execution
if background:
# Run in background with nohup and redirect output
exec_command = f"nohup {command} > /tmp/bg_output.log 2>&1 &"
try:
result = env.execute(exec_command, timeout=10)
return json.dumps({
"output": "Background task started successfully",
"exit_code": 0,
"error": None
}, ensure_ascii=False)
except Exception as e:
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Failed to start background task: {str(e)}"
}, ensure_ascii=False)
else:
# Run foreground command with retry logic
max_retries = 3
retry_count = 0
result = None
while retry_count <= max_retries:
try:
result = env.execute(command, timeout=effective_timeout)
except Exception as e:
error_str = str(e).lower()
if "timeout" in error_str:
return json.dumps({
"output": "",
"exit_code": 124,
"error": f"Command timed out after {effective_timeout} seconds"
}, ensure_ascii=False)
# Retry on transient errors
if retry_count < max_retries:
retry_count += 1
wait_time = 2 ** retry_count
print(f"⚠️ Terminal: execution error, retrying in {wait_time}s (attempt {retry_count}/{max_retries})")
time.sleep(wait_time)
continue
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Command execution failed: {str(e)}"
}, ensure_ascii=False)
# Got a result
break
# Extract output
output = result.get("output", "")
returncode = result.get("returncode", 0)
# Truncate output if too long
MAX_OUTPUT_CHARS = 50000
if len(output) > MAX_OUTPUT_CHARS:
truncated_notice = f"\n\n... [OUTPUT TRUNCATED - showing last {MAX_OUTPUT_CHARS} chars of {len(output)} total] ..."
output = truncated_notice + output[-MAX_OUTPUT_CHARS:]
return json.dumps({
"output": output.strip() if output else "",
"exit_code": returncode,
"error": None
}, ensure_ascii=False)
except Exception as e:
return json.dumps({
"output": "",
"exit_code": -1,
"error": f"Failed to execute command: {str(e)}",
"status": "error"
}, ensure_ascii=False)
def check_terminal_requirements() -> bool:
"""Check if all requirements for the terminal tool are met."""
config = _get_env_config()
env_type = config["env_type"]
try:
if env_type == "local":
from minisweagent.environments.local import LocalEnvironment
return True
elif env_type == "docker":
from minisweagent.environments.docker import DockerEnvironment
# Check if docker is available
import subprocess
result = subprocess.run(["docker", "version"], capture_output=True, timeout=5)
return result.returncode == 0
elif env_type == "modal":
from minisweagent.environments.extra.swerex_modal import SwerexModalEnvironment
# Check for modal token
return os.getenv("MODAL_TOKEN_ID") is not None or Path.home().joinpath(".modal.toml").exists()
else:
return False
except Exception as e:
print(f"Terminal requirements check failed: {e}")
return False
# =============================================================================
# Atropos-Agent Tool wrapper (Hermes compatibility)
# =============================================================================
from .base import Tool, ToolResult, ToolSchema # noqa: E402
from .basic_tools import BashTool # noqa: E402
class TerminalTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="terminal",
description=TERMINAL_TOOL_DESCRIPTION,
parameters={
"command": {"type": "string", "description": "The command to execute"},
"background": {
"type": "boolean",
"description": "Run the command in the background (not supported in sandbox).",
"default": False,
},
"timeout": {
"type": "integer",
"description": "Command timeout in seconds (optional).",
"minimum": 1,
},
},
required=["command"],
)
def is_available(self) -> tuple[bool, str | None]:
# The canonical terminal tool in atropos-agent is sandboxed; availability is governed
# by whether the environment includes this tool and its own network policy.
return True, None
async def execute(
self,
command: str,
background: bool = False,
timeout: Optional[int] = None,
) -> ToolResult:
if background:
return ToolResult(success=False, error="background execution is not supported in sandboxed terminal")
bash = BashTool(timeout=float(timeout) if timeout else 30.0)
return await bash.execute(command=command)
if __name__ == "__main__":
"""Simple test when run directly."""
print("Terminal Tool Module (mini-swe-agent backend)")
print("=" * 50)
config = _get_env_config()
print(f"\nCurrent Configuration:")
print(f" Environment type: {config['env_type']}")
print(f" Docker image: {config['docker_image']}")
print(f" Modal image: {config['modal_image']}")
print(f" Working directory: {config['cwd']}")
print(f" Default timeout: {config['timeout']}s")
print(f" Lifetime: {config['lifetime_seconds']}s")
if not check_terminal_requirements():
print("\n❌ Requirements not met. Please check the messages above.")
exit(1)
print("\n✅ All requirements met!")
print("\nAvailable Tool:")
print(" - terminal_tool: Execute commands using mini-swe-agent environments")
print("\nUsage Examples:")
print(" # Execute a command")
print(" result = terminal_tool(command='ls -la')")
print(" ")
print(" # Run a background task")
print(" result = terminal_tool(command='python server.py', background=True)")
print("\nEnvironment Variables:")
print(f" TERMINAL_ENV: {os.getenv('TERMINAL_ENV', 'local')} (local/docker/modal)")
print(f" TERMINAL_DOCKER_IMAGE: {os.getenv('TERMINAL_DOCKER_IMAGE', 'python:3.11-slim')}")
print(f" TERMINAL_MODAL_IMAGE: {os.getenv('TERMINAL_MODAL_IMAGE', 'python:3.11-slim')}")
print(f" TERMINAL_CWD: {os.getenv('TERMINAL_CWD', '/tmp')}")
print(f" TERMINAL_TIMEOUT: {os.getenv('TERMINAL_TIMEOUT', '60')}")
print(f" TERMINAL_LIFETIME_SECONDS: {os.getenv('TERMINAL_LIFETIME_SECONDS', '300')}")

68
atropos/tools/toolset_distributions.py
View File

@ -1,68 +0,0 @@
"""
Toolset distributions (Hermes-Agent inspired).
Distributions are optional helpers for data generation runs where you want
probabilistic inclusion of toolsets per trajectory/item.
"""
from __future__ import annotations
import random
from typing import Dict, List, Optional, TypedDict
from .toolsets import validate_toolset
class DistributionDef(TypedDict):
description: str
toolsets: Dict[str, int]
DISTRIBUTIONS: Dict[str, DistributionDef] = {
"default": {
"description": "All common sandbox tools.",
"toolsets": {"sandbox": 100},
},
"code_agent_plus_image": {
"description": "Sandbox tools with optional image generation.",
"toolsets": {"sandbox": 100, "image_gen": 30},
},
"sandbox_only": {
"description": "Only sandbox tools (terminal + filesystem).",
"toolsets": {"sandbox": 100},
},
}
def get_distribution(name: str) -> Optional[DistributionDef]:
return DISTRIBUTIONS.get(name)
def list_distributions() -> Dict[str, DistributionDef]:
return DISTRIBUTIONS.copy()
def validate_distribution(name: str) -> bool:
return name in DISTRIBUTIONS
def sample_toolsets_from_distribution(distribution_name: str) -> List[str]:
dist = get_distribution(distribution_name)
if not dist:
raise ValueError(f"Unknown distribution: {distribution_name}")
selected: List[str] = []
for toolset_name, probability in dist["toolsets"].items():
if not validate_toolset(toolset_name):
continue
if random.random() * 100 < probability:
selected.append(toolset_name)
# Ensure at least one toolset if the distribution isn't empty.
if not selected and dist["toolsets"]:
highest = max(dist["toolsets"].items(), key=lambda x: x[1])[0]
if validate_toolset(highest):
selected.append(highest)
return selected

88
atropos/tools/toolset_resolver.py Normal file
View File

@ -0,0 +1,88 @@
"""
Toolset resolution for Hermes-Agent Atropos integration.
We primarily reuse Hermes-Agent toolsets (`toolsets.py`), but Atropos training/envs
need a few extra sandbox-oriented toolsets that Hermes doesn't expose by default
(e.g. filesystem + stateful terminal).
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Set
import toolsets as hermes_toolsets
ATROPOS_TOOLSETS: Dict[str, Dict[str, Any]] = {
"filesystem": {
"description": "Read/write files in the sandbox workspace.",
"tools": ["read_file", "write_file"],
"includes": [],
},
"terminal_stateful": {
"description": "Stateful terminal execution (tmux/TUI support) inside the sandbox.",
"tools": ["terminal_stateful", "tmux"],
"includes": [],
},
"sandbox": {
"description": "Sandbox tools (terminal + filesystem).",
"tools": [],
"includes": ["terminal", "filesystem"],
},
"default": {
"description": "Default toolset for Atropos AgentEnv tasks.",
"tools": [],
"includes": ["sandbox"],
},
"full": {
"description": "All Hermes tools plus Atropos sandbox additions.",
"tools": [],
"includes": ["all", "filesystem", "sandbox", "terminal_stateful"],
},
}
def validate_toolset(name: str) -> bool:
if name in {"all", "*"}:
return True
return hermes_toolsets.validate_toolset(name) or name in ATROPOS_TOOLSETS
def resolve_toolset(name: str, visited: Optional[Set[str]] = None) -> List[str]:
if visited is None:
visited = set()
if name in {"all", "*"}:
# Union Hermes + Atropos toolsets.
all_tools: Set[str] = set()
for tname in hermes_toolsets.get_toolset_names():
all_tools.update(resolve_toolset(tname, visited=set()))
for tname, spec in ATROPOS_TOOLSETS.items():
# Avoid recursion: some Atropos toolsets (e.g. "full") include "all".
if tname == "full" or "all" in (spec.get("includes") or []):
continue
all_tools.update(resolve_toolset(tname, visited=set()))
return sorted(all_tools)
if name in ATROPOS_TOOLSETS:
if name in visited:
return []
visited.add(name)
spec = ATROPOS_TOOLSETS[name]
tools: Set[str] = set(spec.get("tools", []))
for inc in spec.get("includes", []):
tools.update(resolve_toolset(inc, visited=set(visited)))
return sorted(tools)
# Fall back to Hermes toolsets.
# IMPORTANT: do not pre-add `name` to `visited` here; Hermes' resolver uses
# `visited` for its own cycle detection and will treat the presence of `name`
# as a circular dependency.
return sorted(hermes_toolsets.resolve_toolset(name, visited=set(visited)))
def resolve_multiple_toolsets(names: List[str]) -> List[str]:
tools: Set[str] = set()
for name in names:
tools.update(resolve_toolset(name, visited=set()))
return sorted(tools)

157
atropos/tools/toolsets.py
View File

@ -1,157 +0,0 @@
"""
Toolsets (Hermes-Agent inspired).
Toolsets are named groups of tools with optional composition (includes).
They are used to decide which tools are advertised to the model and/or enabled
for a particular environment run.
This module is intentionally lightweight and dependency-free.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Set, TypedDict
class ToolsetDef(TypedDict):
description: str
tools: List[str]
includes: List[str]
TOOLSETS: Dict[str, ToolsetDef] = {
# Primitive building blocks
"filesystem": {
"description": "Read/write files in the current workspace.",
"tools": ["read_file", "write_file"],
"includes": [],
},
"terminal": {
"description": "Terminal/command execution tools.",
# Prefer `terminal` for Hermes compatibility; keep `bash` as a legacy alias.
"tools": ["terminal", "bash"],
"includes": [],
},
"terminal_stateful": {
"description": "Stateful terminal execution (enables persistent background processes like tmux).",
"tools": ["terminal_stateful", "tmux"],
"includes": [],
},
"sandbox": {
"description": "Standard sandbox tools (terminal + filesystem).",
"tools": [],
"includes": ["terminal", "filesystem"],
},
# External tools (executed via ToolServer)
"web": {
"description": "Web research and content extraction tools (external).",
"tools": ["web_search", "web_extract", "web_crawl"],
"includes": [],
},
"vision": {
"description": "Vision/image analysis tools (external).",
"tools": ["vision_analyze"],
"includes": [],
},
"image_gen": {
"description": "Image generation tools (external).",
"tools": ["image_generate"],
"includes": [],
},
"moa": {
"description": "Advanced reasoning tools (Mixture-of-Agents, external).",
"tools": ["mixture_of_agents"],
"includes": [],
},
# Convenience presets
"default": {
"description": "Default toolset for code-agent tasks.",
"tools": [],
"includes": ["sandbox"],
},
"debugging": {
"description": "Debugging toolkit (terminal + web).",
"tools": [],
"includes": ["sandbox", "web"],
},
"research": {
"description": "Research toolkit (web + vision + reasoning).",
"tools": [],
"includes": ["web", "vision", "moa"],
},
"safe": {
"description": "Safe toolkit without terminal access.",
"tools": [],
"includes": ["web", "vision", "image_gen", "moa"],
},
"full": {
"description": "All common tools (sandbox + external).",
"tools": [],
"includes": ["sandbox", "web", "vision", "image_gen", "moa"],
},
}
def get_toolset(name: str) -> Optional[ToolsetDef]:
return TOOLSETS.get(name)
def get_toolset_names() -> List[str]:
return list(TOOLSETS.keys())
def validate_toolset(name: str) -> bool:
return name in {"all", "*"} or name in TOOLSETS
def resolve_toolset(name: str, visited: Optional[Set[str]] = None) -> List[str]:
"""
Recursively resolve a toolset to a list of tool names.
Includes are expanded depth-first with cycle protection.
"""
if visited is None:
visited = set()
if name in {"all", "*"}:
all_tools: Set[str] = set()
for toolset_name in get_toolset_names():
all_tools.update(resolve_toolset(toolset_name, visited=set()))
return sorted(all_tools)
if name in visited:
# Cycle: return empty to avoid infinite recursion.
return []
visited.add(name)
toolset = TOOLSETS.get(name)
if toolset is None:
return []
tools: Set[str] = set(toolset.get("tools", []))
for included in toolset.get("includes", []):
tools.update(resolve_toolset(included, visited=set(visited)))
return sorted(tools)
def resolve_multiple_toolsets(toolset_names: List[str]) -> List[str]:
tools: Set[str] = set()
for name in toolset_names:
tools.update(resolve_toolset(name))
return sorted(tools)
def get_toolset_info(name: str) -> Optional[Dict[str, Any]]:
toolset = get_toolset(name)
if toolset is None:
return None
resolved = resolve_toolset(name)
return {
"name": name,
"description": toolset["description"],
"direct_tools": toolset["tools"],
"includes": toolset["includes"],
"resolved_tools": resolved,
"tool_count": len(resolved),
"is_composite": bool(toolset["includes"]),
}

553
atropos/tools/vision_tools.py
View File

@ -1,553 +0,0 @@
#!/usr/bin/env python3
"""
Vision Tools Module
This module provides vision analysis tools that work with image URLs.
Uses Gemini 3 Flash Preview via OpenRouter API for intelligent image understanding.
Available tools:
- vision_analyze_tool: Analyze images from URLs with custom prompts
Features:
- Downloads images from URLs and converts to base64 for API compatibility
- Comprehensive image description
- Context-aware analysis based on user queries
- Automatic temporary file cleanup
- Proper error handling and validation
- Debug logging support
Usage:
from vision_tools import vision_analyze_tool
import asyncio
# Analyze an image
result = await vision_analyze_tool(
image_url="https://example.com/image.jpg",
user_prompt="What architectural style is this building?"
)
"""
import json
import os
import asyncio
import uuid
import datetime
import base64
from pathlib import Path
from typing import Dict, Any, Optional
from openai import AsyncOpenAI
import httpx # Use httpx for async HTTP requests
# Initialize OpenRouter API client for vision processing
openrouter_client = AsyncOpenAI(
api_key=os.getenv("OPENROUTER_API_KEY"),
base_url="https://openrouter.ai/api/v1"
)
# Configuration for vision processing
DEFAULT_VISION_MODEL = "google/gemini-3-flash-preview"
# Debug mode configuration
DEBUG_MODE = os.getenv("VISION_TOOLS_DEBUG", "false").lower() == "true"
DEBUG_SESSION_ID = str(uuid.uuid4())
DEBUG_LOG_PATH = Path("./logs")
DEBUG_DATA = {
"session_id": DEBUG_SESSION_ID,
"start_time": datetime.datetime.now().isoformat(),
"debug_enabled": DEBUG_MODE,
"tool_calls": []
} if DEBUG_MODE else None
# Create logs directory if debug mode is enabled
if DEBUG_MODE:
DEBUG_LOG_PATH.mkdir(exist_ok=True)
print(f"🐛 Vision debug mode enabled - Session ID: {DEBUG_SESSION_ID}")
def _log_debug_call(tool_name: str, call_data: Dict[str, Any]) -> None:
"""
Log a debug call entry to the global debug data structure.
Args:
tool_name (str): Name of the tool being called
call_data (Dict[str, Any]): Data about the call including parameters and results
"""
if not DEBUG_MODE or not DEBUG_DATA:
return
call_entry = {
"timestamp": datetime.datetime.now().isoformat(),
"tool_name": tool_name,
**call_data
}
DEBUG_DATA["tool_calls"].append(call_entry)
def _save_debug_log() -> None:
"""
Save the current debug data to a JSON file in the logs directory.
"""
if not DEBUG_MODE or not DEBUG_DATA:
return
try:
debug_filename = f"vision_tools_debug_{DEBUG_SESSION_ID}.json"
debug_filepath = DEBUG_LOG_PATH / debug_filename
# Update end time
DEBUG_DATA["end_time"] = datetime.datetime.now().isoformat()
DEBUG_DATA["total_calls"] = len(DEBUG_DATA["tool_calls"])
with open(debug_filepath, 'w', encoding='utf-8') as f:
json.dump(DEBUG_DATA, f, indent=2, ensure_ascii=False)
print(f"🐛 Vision debug log saved: {debug_filepath}")
except Exception as e:
print(f"❌ Error saving vision debug log: {str(e)}")
def _validate_image_url(url: str) -> bool:
"""
Basic validation of image URL format.
Args:
url (str): The URL to validate
Returns:
bool: True if URL appears to be valid, False otherwise
"""
if not url or not isinstance(url, str):
return False
# Check if it's a valid URL format
if not (url.startswith('http://') or url.startswith('https://')):
return False
# Check for common image extensions (optional, as URLs may not have extensions)
image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.svg']
return True # Allow all HTTP/HTTPS URLs for flexibility
async def _download_image(image_url: str, destination: Path, max_retries: int = 3) -> Path:
"""
Download an image from a URL to a local destination (async) with retry logic.
Args:
image_url (str): The URL of the image to download
destination (Path): The path where the image should be saved
max_retries (int): Maximum number of retry attempts (default: 3)
Returns:
Path: The path to the downloaded image
Raises:
Exception: If download fails after all retries
"""
import asyncio
# Create parent directories if they don't exist
destination.parent.mkdir(parents=True, exist_ok=True)
last_error = None
for attempt in range(max_retries):
try:
# Download the image with appropriate headers using async httpx
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(
image_url,
headers={"User-Agent": "hermes-agent-vision/1.0"},
)
response.raise_for_status()
# Save the image content
destination.write_bytes(response.content)
return destination
except Exception as e:
last_error = e
if attempt < max_retries - 1:
wait_time = 2 ** (attempt + 1) # 2s, 4s, 8s
print(f"⚠️ Image download failed (attempt {attempt + 1}/{max_retries}): {str(e)[:50]}")
print(f" Retrying in {wait_time}s...")
await asyncio.sleep(wait_time)
else:
print(f"❌ Image download failed after {max_retries} attempts: {str(e)[:100]}")
raise last_error
def _determine_mime_type(image_path: Path) -> str:
"""
Determine the MIME type of an image based on its file extension.
Args:
image_path (Path): Path to the image file
Returns:
str: The MIME type (defaults to image/jpeg if unknown)
"""
extension = image_path.suffix.lower()
mime_types = {
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
'.gif': 'image/gif',
'.bmp': 'image/bmp',
'.webp': 'image/webp',
'.svg': 'image/svg+xml'
}
return mime_types.get(extension, 'image/jpeg')
def _image_to_base64_data_url(image_path: Path, mime_type: Optional[str] = None) -> str:
"""
Convert an image file to a base64-encoded data URL.
Args:
image_path (Path): Path to the image file
mime_type (Optional[str]): MIME type of the image (auto-detected if None)
Returns:
str: Base64-encoded data URL (e.g., "data:image/jpeg;base64,...")
"""
# Read the image as bytes
data = image_path.read_bytes()
# Encode to base64
encoded = base64.b64encode(data).decode("ascii")
# Determine MIME type
mime = mime_type or _determine_mime_type(image_path)
# Create data URL
data_url = f"data:{mime};base64,{encoded}"
return data_url
async def vision_analyze_tool(
image_url: str,
user_prompt: str,
model: str = DEFAULT_VISION_MODEL
) -> str:
"""
Analyze an image from a URL using vision AI.
This tool downloads images from URLs, converts them to base64, and processes
them using Gemini 3 Flash Preview via OpenRouter API. The image is downloaded to a
temporary location and automatically cleaned up after processing.
The user_prompt parameter is expected to be pre-formatted by the calling
function (typically model_tools.py) to include both full description
requests and specific questions.
Args:
image_url (str): The URL of the image to analyze (must be http:// or https://)
user_prompt (str): The pre-formatted prompt for the vision model
model (str): The vision model to use (default: google/gemini-3-flash-preview)
Returns:
str: JSON string containing the analysis results with the following structure:
{
"success": bool,
"analysis": str (defaults to error message if None)
}
Raises:
Exception: If download fails, analysis fails, or API key is not set
Note:
- Temporary images are stored in ./temp_vision_images/
- Images are automatically deleted after processing
- Supports common image formats (JPEG, PNG, GIF, WebP, etc.)
"""
debug_call_data = {
"parameters": {
"image_url": image_url,
"user_prompt": user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt,
"model": model
},
"error": None,
"success": False,
"analysis_length": 0,
"model_used": model,
"image_size_bytes": 0
}
temp_image_path = None
try:
print(f"🔍 Analyzing image from URL: {image_url[:60]}{'...' if len(image_url) > 60 else ''}", flush=True)
print(f"📝 User prompt: {user_prompt[:100]}{'...' if len(user_prompt) > 100 else ''}", flush=True)
# Validate image URL
if not _validate_image_url(image_url):
raise ValueError("Invalid image URL format. Must start with http:// or https://")
# Check API key availability
if not os.getenv("OPENROUTER_API_KEY"):
raise ValueError("OPENROUTER_API_KEY environment variable not set")
# Download the image to a temporary location
print(f"⬇️ Downloading image from URL...", flush=True)
temp_dir = Path("./temp_vision_images")
temp_image_path = temp_dir / f"temp_image_{uuid.uuid4()}.jpg"
await _download_image(image_url, temp_image_path)
# Get image file size for logging
image_size_bytes = temp_image_path.stat().st_size
image_size_kb = image_size_bytes / 1024
print(f"✅ Image downloaded successfully ({image_size_kb:.1f} KB)", flush=True)
# Convert image to base64 data URL
print(f"🔄 Converting image to base64...", flush=True)
image_data_url = _image_to_base64_data_url(temp_image_path)
# Calculate size in KB for better readability
data_size_kb = len(image_data_url) / 1024
print(f"✅ Image converted to base64 ({data_size_kb:.1f} KB)", flush=True)
debug_call_data["image_size_bytes"] = image_size_bytes
# Use the prompt as provided (model_tools.py now handles full description formatting)
comprehensive_prompt = user_prompt
# Prepare the message with base64-encoded image
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": comprehensive_prompt
},
{
"type": "image_url",
"image_url": {
"url": image_data_url
}
}
]
}
]
print(f"🧠 Processing image with {model}...", flush=True)
# Call the vision API with reasoning enabled
response = await openrouter_client.chat.completions.create(
model=model,
messages=messages,
temperature=0.1, # Low temperature for consistent analysis
max_tokens=2000, # Generous limit for detailed analysis
extra_body={
"reasoning": {
"enabled": True,
"effort": "xhigh"
}
}
)
# Extract the analysis
analysis = response.choices[0].message.content.strip()
analysis_length = len(analysis)
print(f"✅ Image analysis completed ({analysis_length} characters)", flush=True)
# Prepare successful response
result = {
"success": True,
"analysis": analysis or "There was a problem with the request and the image could not be analyzed."
}
debug_call_data["success"] = True
debug_call_data["analysis_length"] = analysis_length
# Log debug information
_log_debug_call("vision_analyze_tool", debug_call_data)
_save_debug_log()
return json.dumps(result, indent=2, ensure_ascii=False)
except Exception as e:
error_msg = f"Error analyzing image: {str(e)}"
print(f"❌ {error_msg}", flush=True)
# Prepare error response
result = {
"success": False,
"analysis": "There was a problem with the request and the image could not be analyzed."
}
debug_call_data["error"] = error_msg
_log_debug_call("vision_analyze_tool", debug_call_data)
_save_debug_log()
return json.dumps(result, indent=2, ensure_ascii=False)
finally:
# Clean up temporary image file
if temp_image_path and temp_image_path.exists():
try:
temp_image_path.unlink()
print(f"🧹 Cleaned up temporary image file", flush=True)
except Exception as cleanup_error:
print(f"⚠️ Warning: Could not delete temporary file: {cleanup_error}", flush=True)
def check_openrouter_api_key() -> bool:
"""
Check if the OpenRouter API key is available in environment variables.
Returns:
bool: True if API key is set, False otherwise
"""
return bool(os.getenv("OPENROUTER_API_KEY"))
def check_vision_requirements() -> bool:
"""
Check if all requirements for vision tools are met.
Returns:
bool: True if requirements are met, False otherwise
"""
return check_openrouter_api_key()
def get_debug_session_info() -> Dict[str, Any]:
"""
Get information about the current debug session.
Returns:
Dict[str, Any]: Dictionary containing debug session information
"""
if not DEBUG_MODE or not DEBUG_DATA:
return {
"enabled": False,
"session_id": None,
"log_path": None,
"total_calls": 0
}
return {
"enabled": True,
"session_id": DEBUG_SESSION_ID,
"log_path": str(DEBUG_LOG_PATH / f"vision_tools_debug_{DEBUG_SESSION_ID}.json"),
"total_calls": len(DEBUG_DATA["tool_calls"])
}
# =============================================================================
# Atropos-Agent Tool wrapper (Hermes compatibility)
# =============================================================================
from .base import Tool, ToolResult, ToolSchema # noqa: E402
def _tool_result_from_json(output: str) -> ToolResult:
try:
data = json.loads(output)
except Exception:
return ToolResult(success=True, output=output)
if isinstance(data, dict) and "success" in data:
ok = bool(data.get("success"))
if not ok:
err = data.get("error") or "Vision analysis failed"
return ToolResult(success=False, output=output, error=str(err))
return ToolResult(success=True, output=output)
return ToolResult(success=True, output=output)
class VisionAnalyzeTool(Tool):
@property
def schema(self) -> ToolSchema:
return ToolSchema(
name="vision_analyze",
description=(
"Analyze images from URLs using AI vision. Provides a comprehensive description and answers a question."
),
parameters={
"image_url": {
"type": "string",
"description": "The URL of the image to analyze (must be publicly accessible HTTP/HTTPS URL)",
},
"question": {
"type": "string",
"description": "A specific question about the image content to answer.",
},
},
required=["image_url", "question"],
external=True,
)
def is_available(self) -> tuple[bool, str | None]:
if not os.getenv("OPENROUTER_API_KEY"):
return False, "OPENROUTER_API_KEY not set"
return True, None
async def execute(self, image_url: str, question: str) -> ToolResult:
full_prompt = (
"Fully describe and explain everything about this image, then answer the following question:\n\n"
f"{question}"
)
output = await vision_analyze_tool(image_url=image_url, user_prompt=full_prompt, model=DEFAULT_VISION_MODEL)
return _tool_result_from_json(output)
if __name__ == "__main__":
"""
Simple test/demo when run directly
"""
print("👁️ Vision Tools Module")
print("=" * 40)
# Check if API key is available
api_available = check_openrouter_api_key()
if not api_available:
print("❌ OPENROUTER_API_KEY environment variable not set")
print("Please set your API key: export OPENROUTER_API_KEY='your-key-here'")
print("Get API key at: https://openrouter.ai/")
exit(1)
else:
print("✅ OpenRouter API key found")
print("🛠️ Vision tools ready for use!")
print(f"🧠 Using model: {DEFAULT_VISION_MODEL}")
# Show debug mode status
if DEBUG_MODE:
print(f"🐛 Debug mode ENABLED - Session ID: {DEBUG_SESSION_ID}")
print(f" Debug logs will be saved to: ./logs/vision_tools_debug_{DEBUG_SESSION_ID}.json")
else:
print("🐛 Debug mode disabled (set VISION_TOOLS_DEBUG=true to enable)")
print("\nBasic usage:")
print(" from vision_tools import vision_analyze_tool")
print(" import asyncio")
print("")
print(" async def main():")
print(" result = await vision_analyze_tool(")
print(" image_url='https://example.com/image.jpg',")
print(" user_prompt='What do you see in this image?'")
print(" )")
print(" print(result)")
print(" asyncio.run(main())")
print("\nExample prompts:")
print(" - 'What architectural style is this building?'")
print(" - 'Describe the emotions and mood in this image'")
print(" - 'What text can you read in this image?'")
print(" - 'Identify any safety hazards visible'")
print(" - 'What products or brands are shown?'")
print("\nDebug mode:")
print(" # Enable debug logging")
print(" export VISION_TOOLS_DEBUG=true")
print(" # Debug logs capture all vision analysis calls and results")
print(" # Logs saved to: ./logs/vision_tools_debug_UUID.json")

1417
atropos/tools/web_tools.py
View File

File diff suppressed because it is too large Load Diff