Call an agent by name Args: name: Agent name input: Input for the agent ctx: Agent context Returns: Agent result
(self, name: str, input: Dict[str, Any], ctx: SessionContext = None, **kwargs)
| 1252 | logger.error(f"| ❌ Error during agent context manager cleanup: {e}") |
| 1253 | |
| 1254 | async def __call__(self, name: str, input: Dict[str, Any], ctx: SessionContext = None, **kwargs) -> Any: |
| 1255 | """Call an agent by name |
| 1256 | |
| 1257 | Args: |
| 1258 | name: Agent name |
| 1259 | input: Input for the agent |
| 1260 | ctx: Agent context |
| 1261 | Returns: |
| 1262 | Agent result |
| 1263 | """ |
| 1264 | if ctx is None: |
| 1265 | ctx = SessionContext() |
| 1266 | |
| 1267 | agent_info = await self.get_info(name) |
| 1268 | |
| 1269 | # Agent args |
| 1270 | agent_args = { |
| 1271 | "ctx": ctx, |
| 1272 | } |
| 1273 | |
| 1274 | version = agent_info.version |
| 1275 | agent_instance = agent_info.instance |
| 1276 | logger.info(f"| ✅ Using agent {name}@{version}") |
| 1277 | |
| 1278 | return await agent_instance(**input, **agent_args) |
| 1279 |
nothing calls this directly
no test coverage detected