06eeea3dee
Some checks failed
Invert the agent model: the agent IS the computer. The message pump becomes the kernel, handlers are sandboxed apps, and all access is mediated by the platform. Phase 1 — Container foundation: - Multi-stage Dockerfile (python:3.12-slim, non-root user, /data volume) - deploy/entrypoint.py with --dry-run config validation - docker-compose.yml (cap_drop ALL, read_only, no-new-privileges) - docker-compose.dev.yml overlay for development - CI Docker build smoke test Phase 2 — Security hardening: - xml_pipeline/security/ module with default-deny container mode - Permission gate: per-listener tool allowlist enforcement - Network policy: egress control (only declared LLM backend domains) - Shell tool: disabled in container mode - File tool: restricted to /data and /config in container mode - Fetch tool: integrates network egress policy - Config loader: parses security and network YAML sections Phase 3 — Management plane: - Agent app (port 8080): minimal /health, /inject, /ws only - Management app (port 9090): full API, audit log, dashboard - SQLite-backed audit log for tool invocations and security events - Static web dashboard (no framework, WebSocket-driven) - CLI --split flag for dual-port serving All 439 existing tests pass with zero regressions. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
129 lines
3.5 KiB
Python
129 lines
3.5 KiB
Python
"""
|
|
Base classes and registry for tools.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
from functools import wraps
|
|
import inspect
|
|
|
|
|
|
@dataclass
|
|
class ToolResult:
|
|
"""Result from a tool invocation."""
|
|
success: bool
|
|
data: Any = None
|
|
error: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class Tool:
|
|
"""Tool metadata and implementation."""
|
|
name: str
|
|
description: str
|
|
func: Callable
|
|
parameters: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
async def invoke(self, *, _listener_name: Optional[str] = None, **kwargs) -> ToolResult:
|
|
"""
|
|
Invoke the tool with given parameters.
|
|
|
|
Args:
|
|
_listener_name: The invoking listener's name (for permission checks).
|
|
Prefixed with _ to avoid collision with tool parameters.
|
|
**kwargs: Tool-specific parameters.
|
|
"""
|
|
# Permission gate check (container mode)
|
|
if _listener_name is not None:
|
|
from xml_pipeline.tools.permission_gate import check_permission
|
|
denied = check_permission(_listener_name, self.name)
|
|
if denied is not None:
|
|
return denied
|
|
|
|
try:
|
|
result = await self.func(**kwargs)
|
|
if isinstance(result, ToolResult):
|
|
return result
|
|
return ToolResult(success=True, data=result)
|
|
except Exception as e:
|
|
return ToolResult(success=False, error=str(e))
|
|
|
|
|
|
class ToolRegistry:
|
|
"""Registry of available tools."""
|
|
|
|
def __init__(self):
|
|
self._tools: Dict[str, Tool] = {}
|
|
|
|
def register(self, tool: Tool):
|
|
"""Register a tool."""
|
|
self._tools[tool.name] = tool
|
|
|
|
def get(self, name: str) -> Optional[Tool]:
|
|
"""Get a tool by name."""
|
|
return self._tools.get(name)
|
|
|
|
def list(self) -> List[str]:
|
|
"""List all registered tool names."""
|
|
return list(self._tools.keys())
|
|
|
|
def all(self) -> Dict[str, Tool]:
|
|
"""Get all tools."""
|
|
return dict(self._tools)
|
|
|
|
|
|
# Global registry
|
|
_registry: Optional[ToolRegistry] = None
|
|
|
|
|
|
def get_tool_registry() -> ToolRegistry:
|
|
"""Get the global tool registry."""
|
|
global _registry
|
|
if _registry is None:
|
|
_registry = ToolRegistry()
|
|
return _registry
|
|
|
|
|
|
def tool(func: Callable) -> Callable:
|
|
"""Decorator to register a function as a tool."""
|
|
# Extract metadata from function
|
|
name = func.__name__
|
|
description = func.__doc__ or ""
|
|
|
|
# Extract parameters from signature
|
|
sig = inspect.signature(func)
|
|
parameters = {}
|
|
for param_name, param in sig.parameters.items():
|
|
param_info = {"name": param_name}
|
|
if param.annotation != inspect.Parameter.empty:
|
|
ann = param.annotation
|
|
# Handle both string annotations (from __future__ import annotations) and type objects
|
|
if isinstance(ann, str):
|
|
param_info["type"] = ann
|
|
elif hasattr(ann, "__name__"):
|
|
param_info["type"] = ann.__name__
|
|
else:
|
|
param_info["type"] = str(ann)
|
|
if param.default != inspect.Parameter.empty:
|
|
param_info["default"] = param.default
|
|
parameters[param_name] = param_info
|
|
|
|
# Create tool
|
|
t = Tool(
|
|
name=name,
|
|
description=description.strip(),
|
|
func=func,
|
|
parameters=parameters,
|
|
)
|
|
|
|
# Register
|
|
get_tool_registry().register(t)
|
|
|
|
@wraps(func)
|
|
async def wrapper(**kwargs):
|
|
return await t.invoke(**kwargs)
|
|
|
|
wrapper._tool = t
|
|
return wrapper
|