Execute all tool_use blocks in one Anthropic response turn.
(
ctx: _AgentRunContext,
response_content: list[Any],
)
| 417 | |
| 418 | |
| 419 | async def _process_tool_turn( |
| 420 | ctx: _AgentRunContext, |
| 421 | response_content: list[Any], |
| 422 | ) -> None: |
| 423 | """Execute all tool_use blocks in one Anthropic response turn.""" |
| 424 | tool_results = [] |
| 425 | for block in response_content: |
| 426 | if block.type != "tool_use": |
| 427 | continue |
| 428 | result = await _execute_tool(block.name, block.input, ctx.on_tool_call) |
| 429 | ctx.tool_calls.append(ToolCall(name=block.name, input=block.input, result=result)) |
| 430 | logger.info("Tool executed: %s → %d chars", block.name, len(result)) |
| 431 | tool_results.append( |
| 432 | { |
| 433 | "type": "tool_result", |
| 434 | "tool_use_id": block.id, |
| 435 | "content": result, |
| 436 | } |
| 437 | ) |
| 438 | ctx.messages.append({"role": "user", "content": tool_results}) |
| 439 | |
| 440 | |
| 441 | def _build_ollama_assistant_message(msg: Any) -> dict[str, Any]: |
no test coverage detected