« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""AgentAsToolAdapter — wraps any agent as a ToolProtocol.
3Enables hierarchical multi-agent delegation by exposing an agent as
4a tool that other agents can invoke through their normal tool-calling
5reasoning loop. The supervisor agent sees delegated agents as tools
6with descriptive names and schemas, and the adapter transparently
7routes execution through the ``AgentExecutorProtocol``.
9Example::
11 from lexigram.ai.agents.delegation import AgentAsToolAdapter
13 billing_tool = AgentAsToolAdapter(
14 agent=billing_agent,
15 executor=executor,
16 )
17 # Use as any other tool in a supervisor's tool list
18 supervisor = Agent(
19 name="supervisor",
20 tools=[billing_tool, technical_tool],
21 strategy=ReActStrategy(),
22 )
23"""
25from __future__ import annotations
27from typing import TYPE_CHECKING, Any
29from lexigram.logging import (
30 get_logger,
31)
33if TYPE_CHECKING:
34 from lexigram.contracts.ai.agents import (
35 AgentExecutorProtocol,
36 AgentProtocol,
37 )
39logger = get_logger(__name__)
41_DESCRIPTION_MAX_CHARS = 200
44class AgentAsToolAdapter:
45 """Wraps an ``AgentProtocol`` as a ``ToolProtocol``.
47 The adapter satisfies the ``ToolProtocol`` contract so that any
48 agent can be injected into another agent's tool list. When
49 ``execute()`` is called, the adapter delegates to the provided
50 ``AgentExecutorProtocol.run()`` with the given message.
52 Attributes:
53 name: ``delegate_to_{agent.name}`` — uniquely identifies this
54 delegation tool.
55 description: Derived from the wrapped agent's system prompt,
56 truncated to ``_DESCRIPTION_MAX_CHARS``.
57 parameters_schema: Accepts a single ``message`` string.
58 """
60 def __init__(
61 self,
62 agent: AgentProtocol,
63 executor: AgentExecutorProtocol,
64 *,
65 session_id: str | None = None,
66 user_id: str | None = None,
67 ) -> None:
68 """Initialize the agent-to-tool adapter.
70 Args:
71 agent: The agent to expose as a tool.
72 executor: The executor used to run the agent.
73 session_id: Optional session ID to pass through to the executor.
74 user_id: Optional user ID for governance tracking.
75 """
76 self._agent = agent
77 self._executor = executor
78 self._session_id = session_id
79 self._user_id = user_id
81 @property
82 def name(self) -> str:
83 """Unique tool identifier derived from the wrapped agent name."""
84 return f"delegate_to_{self._agent.name}"
86 @property
87 def description(self) -> str:
88 """Human-readable description for LLM tool selection."""
89 prompt = self._agent.system_prompt or ""
90 truncated = prompt[:_DESCRIPTION_MAX_CHARS]
91 if len(prompt) > _DESCRIPTION_MAX_CHARS:
92 truncated += "…"
93 return f"Delegate task to '{self._agent.name}': {truncated}"
95 @property
96 def parameters_schema(self) -> dict[str, Any]:
97 """JSON Schema for the delegation tool parameters."""
98 return {
99 "type": "object",
100 "properties": {
101 "message": {
102 "type": "string",
103 "description": (
104 f"The task or question to delegate to the "
105 f"'{self._agent.name}' agent."
106 ),
107 },
108 },
109 "required": ["message"],
110 }
112 async def execute(self, **kwargs: Any) -> Any:
113 """Execute by delegating to the wrapped agent.
115 Args:
116 **kwargs: Must contain ``message`` (str) — the task to delegate.
118 Returns:
119 The agent's response message string on success, or an error
120 description string on failure.
121 """
122 message = kwargs.get("message", "")
123 if not message:
124 return "Error: No message provided for delegation."
126 logger.info(
127 "agent_delegation_start",
128 from_tool=self.name,
129 to_agent=self._agent.name,
130 message_length=len(message),
131 )
133 result = await self._executor.run(
134 agent=self._agent,
135 message=message,
136 session_id=self._session_id,
137 user_id=self._user_id,
138 )
140 if result.is_ok():
141 response = result.unwrap()
142 logger.info(
143 "agent_delegation_complete",
144 to_agent=self._agent.name,
145 steps=response.step_count,
146 tokens=response.total_tokens,
147 )
148 return response.message
150 error = result.unwrap_err()
151 logger.warning(
152 "agent_delegation_failed",
153 to_agent=self._agent.name,
154 error=str(error),
155 )
156 return f"Delegation to '{self._agent.name}' failed: {error}"
159__all__ = ["AgentAsToolAdapter"]