Skip to content

tool-discovery.md

Latest commit

 

History

History
149 lines (107 loc) · 4.99 KB

tool-discovery.md

File metadata and controls

149 lines (107 loc) · 4.99 KB

Tool Discovery, Sandbox, and Approval

The ECS Agent framework provides a complete pipeline for tool management: automatic discovery, sandboxed execution, and human-in-the-loop approval.

Enhanced Events

The framework features an extensive event system for monitoring tool execution, skill lifecycles, and MCP connectivity. These events are published to the EventBus and can be used for logging, UI updates, or debugging.

New Event Types

Event Name Fired When Key Fields
ToolExecutionStartedEvent Before a tool handler is called tool_call, entity_id
ToolExecutionCompletedEvent After a tool handler returns tool_call_id, result, success
SkillInstalledEvent After a skill is installed skill_name, tool_names, entity_id
SkillUninstalledEvent After a skill is uninstalled skill_name, entity_id
SkillDiscoveryEvent After scanning a discovery source source, skills_found, errors
MCPConnectedEvent After an MCP server connects server_name
MCPDisconnectedEvent After an MCP server disconnects server_name
MCPToolCallEvent After an MCP tool is called server_name, tool_name, success

Code Example: Subscribing to Events

You can subscribe to these events to track the agent's progress in real-time.

async def on_tool_start(event: ToolExecutionStartedEvent):
    print(f"Tool {event.tool_call.name} started execution.")

async def on_discovery(event: SkillDiscoveryEvent):
    print(f"Discovered {len(event.skills_found)} skills from {event.source}")

world.event_bus.subscribe(ToolExecutionStartedEvent, on_tool_start)
world.event_bus.subscribe(SkillDiscoveryEvent, on_discovery)

Tool Auto-Discovery

The scan_module function automatically discovers tool-decorated functions in a Python module.

The @tool Decorator

from ecs_agent.tools import tool

@tool(
    name="calculate",
    description="Perform arithmetic calculations",
    parameters={
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "Math expression to evaluate"}
        },
        "required": ["expression"]
    }
)
async def calculate(expression: str) -> str:
    return str(eval(expression))

Scanning a Module

from ecs_agent.tools import scan_module
import my_tools_module

tools, handlers = scan_module(my_tools_module)
# tools: dict[str, ToolSchema] — tool name to schema mapping
# handlers: dict[str, Callable] — tool name to async handler mapping

world.add_component(agent, ToolRegistryComponent(tools=tools, handlers=handlers))

Sandboxed Execution

The sandboxed_execute function runs tool handlers with timeout and output size limits.

from ecs_agent.tools import sandboxed_execute

result = await sandboxed_execute(
    func=my_handler,
    args={"expression": "2 + 2"},
    timeout=30.0,         # Maximum execution time in seconds
    max_output_size=10000  # Maximum output size in bytes
)

Configuration via SandboxConfigComponent

from ecs_agent.components import SandboxConfigComponent

world.add_component(agent, SandboxConfigComponent(timeout=10.0, max_output_size=5000))

Tool Approval System

The ToolApprovalSystem provides human-in-the-loop approval for tool calls before execution.

Approval Policies

from ecs_agent.types import ApprovalPolicy

# Three policies available:
ApprovalPolicy.ALWAYS_APPROVE   # All tool calls pass through
ApprovalPolicy.ALWAYS_DENY      # All tool calls are blocked
ApprovalPolicy.REQUIRE_APPROVAL # Human must approve each call

Setup

from ecs_agent.components import ToolApprovalComponent
from ecs_agent.systems.tool_approval import ToolApprovalSystem

world.add_component(agent, ToolApprovalComponent(
    policy=ApprovalPolicy.REQUIRE_APPROVAL,
    timeout=60.0,  # Seconds to wait for approval; None for infinite wait
))
world.register_system(ToolApprovalSystem(priority=-5), priority=-5)

Approval Flow

  1. LLM generates tool calls (via ReasoningSystem).
  2. ToolApprovalSystem intercepts pending calls before ToolExecutionSystem.
  3. For REQUIRE_APPROVAL policy:
    • A ToolApprovalRequestedEvent is published with a Future[bool].
    • External code resolves the future: event.future.set_result(True) to approve, False to deny.
    • On timeout (or ToolTimeoutError), the call is denied.
  4. Approved calls proceed to ToolExecutionSystem; denied calls are removed.

Events

  • ToolApprovalRequestedEvent(entity_id, tool_call, future) — Requesting approval
  • ToolApprovedEvent(entity_id, tool_call_id) — Call was approved
  • ToolDeniedEvent(entity_id, tool_call_id) — Call was denied

Priority Order

The approval system must run before tool execution:

  • ToolApprovalSystem at priority -5
  • ToolExecutionSystem at priority 5

Complete Example

See examples/tool_approval_agent.py for a full working demo.