Tool Loop Orchestration

The ProductionAgent class wraps the adapter with safety limits, parallel execution, and per-tool timeouts.

Two layers, two responsibilities

The Adapter knows the wire: how to format messages for this provider, how to parse this provider's events, how to send tool results in this provider's shape. The Orchestrator knows the loop: when to call a handler, when to log to JSONL, when to break, when to apply hooks.

Putting the loop inside the Adapter is the most common architectural mistake. It feels simpler ('the adapter handles everything!') but it makes per-call instrumentation impossible. You can't add a logging hook for a single tool call. You can't intercept arguments before the handler runs. You can't replay-test the loop independently of the wire. Keep them separate.

Code

Orchestrator that owns the loop, not the adapter·python

import asyncio, json, logging
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class AgentConfig:
    model: str = "gpt-4.1"
    max_iterations: int = 10        # Safety limit on tool loops
    tool_timeout_seconds: float = 30.0

class ProductionAgent:
    def __init__(self, adapter, tools, tool_schemas, config=None):
        self.adapter = adapter
        self.tools = tools            # name → callable
        self.tool_schemas = tool_schemas
        self.config = config or AgentConfig()

    async def run(self, user_message, conversation_history=None):
        messages = conversation_history or [
            {"role": "system", "content": "You are a helpful assistant."}
        ]
        messages.append({"role": "user", "content": user_message})

        for iteration in range(self.config.max_iterations):
            text, tool_calls = "", []
            async for chunk in self.adapter.stream(
                messages=messages, tools=self.tool_schemas
            ):
                if isinstance(chunk, TextDelta): text += chunk.content
                elif isinstance(chunk, ToolCallComplete):
                    tool_calls.append(chunk)

            if not tool_calls:
                messages.append({"role": "assistant", "content": text})
                return text, messages

            # Execute tools in parallel with per-tool timeout
            results = await asyncio.gather(*[
                self._execute_tool(tc) for tc in tool_calls
            ])
            # Add results to messages...

        return "Max iterations reached.", messages

    async def _execute_tool(self, tc):
        try:
            fn = self.tools[tc.name]
            args = json.loads(tc.arguments)
            return await asyncio.wait_for(fn(**args), timeout=self.config.tool_timeout_seconds)
        except asyncio.TimeoutError:
            return {"error": f"Tool {tc.name} timed out"}
        except Exception as e:
            return {"error": str(e)}

Two layers, two responsibilities

Code

External links

Exercise

Progress

Comments 0