3.6: Add delegate_to_agent tool for cross-agent delegation

New tool in delegate.py routes tasks to named agents via
agent.respond_to_prompt(). Includes thread-local depth counter
(max 3) to prevent infinite A->B->A delegation loops. Extended
ctx injection in ToolRegistry to include agent_registry. Wired
agent_registry into ToolRegistry from __main__.py.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
cora-start
PeninsulaInd 2026-02-17 10:13:32 -06:00
parent 9d4d12e232
commit e5e9442e3d
3 changed files with 59 additions and 5 deletions

View File

@ -104,10 +104,11 @@ def main():
agent_cfg.memory_scope or "shared",
)
# Update tool registry to reference the default agent (for ctx injection)
# Update tool registry to reference the default agent and agent registry
default_agent = registry.default
if tools and default_agent:
tools.agent = default_agent
tools.agent_registry = registry
# Notification bus (UI-agnostic)
notification_bus = None

View File

@ -104,6 +104,7 @@ class ToolRegistry:
self.config = config
self.db = db
self.agent = agent
self.agent_registry = None # set after multi-agent setup
self._discover_tools()
def _discover_tools(self):
@ -156,6 +157,7 @@ class ToolRegistry:
"db": self.db,
"agent": self.agent,
"memory": self.agent._memory,
"agent_registry": self.agent_registry,
}
result = tool_def.func(**args)
return str(result) if result is not None else "Done."

View File

@ -1,14 +1,22 @@
"""Delegate tool: bridges chat brain to execution brain.
"""Delegate tools: bridges between brains and between agents.
When the chat model needs to run commands, edit files, or do anything
requiring system-level access, it calls this tool. The task is passed
to the execution brain (Claude Code CLI) which has full tool access.
delegate_task sends a task to the execution brain (Claude Code CLI).
delegate_to_agent routes a task to a named agent in the multi-agent registry.
"""
from __future__ import annotations
import logging
import threading
from . import tool
log = logging.getLogger(__name__)
# Guard against infinite agent-to-agent delegation loops.
_MAX_DELEGATION_DEPTH = 3
_delegation_depth = threading.local()
@tool(
"delegate_task",
@ -28,3 +36,46 @@ def delegate_task(task_description: str, ctx: dict | None = None) -> str:
agent = ctx["agent"]
return agent.execute_task(task_description)
@tool(
"delegate_to_agent",
description=(
"Route a task to a specific named agent. Use this to delegate work to "
"a specialist: e.g. 'researcher' for deep research, 'writer' for content "
"creation, 'ops' for system operations. The target agent processes the "
"task using its own tools, skills, and memory scope, then returns the result."
),
category="system",
)
def delegate_to_agent(
agent_name: str, task_description: str, ctx: dict | None = None
) -> str:
"""Delegate a task to another agent by name."""
if not ctx or "agent_registry" not in ctx:
return "Error: delegate_to_agent requires agent_registry in context."
# Depth guard — prevent infinite A→B→A loops
depth = getattr(_delegation_depth, "value", 0)
if depth >= _MAX_DELEGATION_DEPTH:
return (
f"Error: delegation depth limit ({_MAX_DELEGATION_DEPTH}) reached. "
"Cannot delegate further to prevent infinite loops."
)
registry = ctx["agent_registry"]
target = registry.get(agent_name)
if target is None:
available = ", ".join(registry.list_agents())
return f"Error: agent '{agent_name}' not found. Available agents: {available}"
log.info(
"Delegating to agent '%s' (depth %d): %s",
agent_name, depth + 1, task_description[:100],
)
_delegation_depth.value = depth + 1
try:
return target.respond_to_prompt(task_description)
finally:
_delegation_depth.value = depth