82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
"""Delegate tools: bridges between brains and between agents.
|
|
|
|
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",
|
|
description=(
|
|
"Delegate a complex task to the execution brain (Claude Code CLI). "
|
|
"Use this when you need to: run shell commands, read/write/edit files, "
|
|
"check system status, inspect the codebase, or perform any system-level "
|
|
"operation. Describe the task clearly and the execution brain will carry "
|
|
"it out using its full tool suite (Bash, Read, Edit, Write, Glob, Grep)."
|
|
),
|
|
category="system",
|
|
)
|
|
def delegate_task(task_description: str, ctx: dict | None = None) -> str:
|
|
"""Delegate a task to the execution brain."""
|
|
if not ctx or "agent" not in ctx:
|
|
return "Error: delegate tool requires agent context."
|
|
|
|
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
|