""" Hermes Plugin for Agentsd ========================= Bridges Hermes Agent's tool registry into agentsd as a standard plugin. All Hermes tools become callable via agentsd's gRPC PluginBridge. Architecture: agentsd (Rust) ←gRPC→ hermes-plugin.py ← Hermes tools/registry.py This plugin: 1. Imports Hermes tool registry and discovers all built-in tools 2. Registers with agentsd as a "basic" plugin, exporting all Hermes tools 3. Listens for Call requests from agentsd and dispatches to Hermes handlers """ import asyncio import json import logging import os import sys import threading import time from concurrent import futures from pathlib import Path import grpc from grpc import StatusCode # Add hermes-agent to path HERMES_DIR = Path(__file__).resolve().parent.parent.parent.parent / "hermes-agent" sys.path.insert(0, str(HERMES_DIR)) # Import generated proto SDK_DIR = Path(__file__).resolve().parent.parent / "_sdk" sys.path.insert(0, str(SDK_DIR)) import agentsd_pb2 as pb import agentsd_pb2_grpc as rpc logging.basicConfig(level=logging.INFO, format="[hermes-plugin] %(levelname)s %(message)s") logger = logging.getLogger(__name__) class HermesPlugin: """Bridges Hermes tools into agentsd.""" def __init__(self, agentsd_addr: str = None): # agentsd binds [::1]:50051 (IPv6 loopback); match it so "localhost" # resolving to IPv4 first doesn't break the connection. AGENTSD_ADDR overrides. self.addr = agentsd_addr or os.environ.get("AGENTSD_ADDR", "[::1]:50051") self.channel = None self.bridge = None self.registry = None self.tool_handlers = {} def load_hermes_tools(self): """Import and discover all Hermes built-in tools.""" logger.info("Loading Hermes tools from: %s", HERMES_DIR) # Import registry from tools.registry import registry, discover_builtin_tools # Discover and import all tool modules imported = discover_builtin_tools(HERMES_DIR / "tools") logger.info("Imported %d tool modules", len(imported)) self.registry = registry # Collect all tool names and handlers all_names = registry.get_all_tool_names() for name in all_names: entry = registry.get_entry(name) if entry: self.tool_handlers[name] = entry logger.info("Loaded %d tools: %s", len(self.tool_handlers), list(self.tool_handlers.keys())[:10]) return all_names def connect(self): """Connect to agentsd gRPC server.""" self.channel = grpc.insecure_channel(self.addr) self.bridge = rpc.PluginBridgeStub(self.channel) logger.info("Connected to agentsd at %s", self.addr) def register(self, tool_names): """Register this plugin with agentsd, exporting all Hermes tools.""" exports = [] for name in tool_names: entry = self.tool_handlers.get(name) desc = entry.description if entry else "" exports.append(pb.ExportDef(name=name, description=desc[:200])) resp = self.bridge.Register(pb.PluginInfo( id="hermes", name="Hermes Agent Tools", version="0.16.0", plugin_type="basic", syscalls=["network.request", "fs.read", "fs.write", "process.spawn", "memory.read", "memory.write", "display.text", "timer.once"], depends=[], exports=exports, )) if resp.ok: logger.info("Registered with agentsd: session=%s, %d exports", resp.session_id, len(exports)) else: logger.error("Registration failed: %s", resp.error) return resp.ok def call_tool(self, tool_name: str, args: dict) -> dict: """Execute a Hermes tool by name with given args.""" entry = self.tool_handlers.get(tool_name) if not entry: return {"error": f"tool not found: {tool_name}"} try: if entry.is_async: # Run async handler in event loop loop = asyncio.new_event_loop() try: result = loop.run_until_complete(entry.handler(args)) finally: loop.close() else: result = entry.handler(args) # Hermes tools return strings or dicts if isinstance(result, str): return {"result": result} elif isinstance(result, dict): return result else: return {"result": str(result)} except Exception as e: logger.error("Tool %s failed: %s", tool_name, e) return {"error": str(e)} def run_server(self, port: int = 50052): """ Run a gRPC server that agentsd can call back into. Agentsd routes Call requests here for Hermes tools. """ from concurrent import futures server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) rpc.add_PluginBridgeServicer_to_server(HermesServicer(self), server) server.add_insecure_port(f"[::]:{port}") server.start() logger.info("Hermes plugin server listening on port %d", port) return server class HermesServicer(rpc.PluginBridgeServicer): """gRPC servicer that handles tool call requests from agentsd.""" def __init__(self, plugin: HermesPlugin): self.plugin = plugin def Call(self, request, context): """Handle a tool call from agentsd.""" fn_name = request.fn try: args = json.loads(request.args) if request.args else {} except json.JSONDecodeError: args = {} logger.info("Call: %s(%s)", fn_name, list(args.keys())) result = self.plugin.call_tool(fn_name, args) result_bytes = json.dumps(result, ensure_ascii=False, default=str).encode("utf-8") if "error" in result and not result.get("result"): return pb.CallResponse(result=result_bytes, ok=False, error=result["error"]) return pb.CallResponse(result=result_bytes, ok=True, error="") def Register(self, request, context): context.set_code(StatusCode.UNIMPLEMENTED) return pb.RegisterResponse(ok=False, error="not a bridge") def Heartbeat(self, request, context): return pb.HeartbeatResponse(ok=True) def StreamCall(self, request, context): context.set_code(StatusCode.UNIMPLEMENTED) return pb.CallResponse(ok=False, error="not implemented") def main(): plugin = HermesPlugin() # Load all Hermes tools try: tool_names = plugin.load_hermes_tools() except Exception as e: logger.error("Failed to load Hermes tools: %s", e) logger.info("Ensure hermes-agent is at: %s", HERMES_DIR) sys.exit(1) # Connect and register with agentsd plugin.connect() if not plugin.register(tool_names): sys.exit(1) # Start callback server for tool invocations server = plugin.run_server(port=50052) logger.info("Hermes plugin running. %d tools available. Ctrl+C to stop.", len(tool_names)) try: while True: time.sleep(1) except KeyboardInterrupt: logger.info("Shutting down...") server.stop(grace=5) plugin.channel.close() if __name__ == "__main__": main()