134 lines
3.8 KiB
Python
134 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Pi Agent Tool - Integrate Pi agent with Hermes
|
|
|
|
This tool allows Hermes to delegate tasks to a Pi agent running
|
|
as an HTTP server.
|
|
|
|
Flow:
|
|
Hermes Agent → pi_agent_tool → HTTP Server (Level 4) → Pi Agent
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import requests
|
|
from typing import Any, Dict, Optional
|
|
|
|
# Configuration
|
|
PI_SERVER_URL = os.environ.get("PI_SERVER_URL", "http://localhost:3000")
|
|
PI_TIMEOUT = int(os.environ.get("PI_TIMEOUT", "300"))
|
|
|
|
|
|
def check_pi_requirements() -> bool:
|
|
"""Check if Pi server is available."""
|
|
try:
|
|
response = requests.get(f"{PI_SERVER_URL}/health", timeout=5)
|
|
return response.status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def pi_agent_tool(
|
|
message: str,
|
|
context: Optional[str] = None,
|
|
max_iterations: Optional[int] = None,
|
|
) -> str:
|
|
"""
|
|
Delegate a task to the Pi agent.
|
|
|
|
Args:
|
|
message: The task/message to send to the Pi agent
|
|
context: Optional context to prepend
|
|
max_iterations: Max agent turns (optional)
|
|
|
|
Returns:
|
|
The agent's response
|
|
"""
|
|
# Build the full message with context
|
|
full_message = message
|
|
if context:
|
|
full_message = f"{context}\n\nTask: {message}"
|
|
|
|
try:
|
|
# Call the Pi server
|
|
response = requests.post(
|
|
f"{PI_SERVER_URL}/message",
|
|
json={
|
|
"message": full_message,
|
|
"max_iterations": max_iterations,
|
|
},
|
|
timeout=PI_TIMEOUT,
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return data.get("response", "No response")
|
|
else:
|
|
return f"Error: Server returned {response.status_code}"
|
|
|
|
except requests.Timeout:
|
|
return "Error: Pi agent timed out"
|
|
except requests.ConnectionError:
|
|
return "Error: Cannot connect to Pi server. Is it running?"
|
|
except Exception as e:
|
|
return f"Error: {str(e)}"
|
|
|
|
|
|
# =============================================================================
|
|
# OpenAI Function-Calling Schema
|
|
# =============================================================================
|
|
|
|
PI_AGENT_SCHEMA = {
|
|
"name": "pi_agent",
|
|
"description": (
|
|
"Delegate a coding task to the Pi agent. "
|
|
"Use this for: "
|
|
"1. Complex multi-step tasks "
|
|
"2. Tasks requiring file operations "
|
|
"3. Tasks requiring shell commands "
|
|
"4. Research or investigation tasks "
|
|
"The Pi agent has access to terminal, file operations, and web search.\n\n"
|
|
"Returns the agent's full response."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"message": {
|
|
"type": "string",
|
|
"description": "The task or question to delegate to the Pi agent"
|
|
},
|
|
"context": {
|
|
"type": "string",
|
|
"description": (
|
|
"Optional context to provide to the agent. "
|
|
"Include relevant files, code snippets, or background info."
|
|
)
|
|
},
|
|
"max_iterations": {
|
|
"type": "integer",
|
|
"description": "Maximum number of agent turns (default: 50)"
|
|
}
|
|
},
|
|
"required": ["message"]
|
|
}
|
|
}
|
|
|
|
|
|
# =============================================================================
|
|
# Registry
|
|
# =============================================================================
|
|
from tools.registry import registry, tool_error
|
|
|
|
registry.register(
|
|
name="pi_agent",
|
|
toolset="pi_agent",
|
|
schema=PI_AGENT_SCHEMA,
|
|
handler=lambda args, **kw: pi_agent_tool(
|
|
message=args.get("message"),
|
|
context=args.get("context"),
|
|
max_iterations=args.get("max_iterations"),
|
|
),
|
|
check_fn=check_pi_requirements,
|
|
emoji="🤖",
|
|
)
|