Implement the agent loop

Define a small adapter contract

Do not let an SDK’s provider-specific response object spread through the whole application. Normalize it at one boundary into either a final answer or a tool request. Your own loop can then be unit-tested without a network connection.

# One normalized turn from your model adapter
{"type": "tool_call", "name": "count_words",
 "arguments": {"text": "agent harness"}}

# Or a completed turn
{"type": "final", "text": "The phrase has 2 words."}

In a real provider integration, the adapter translates the provider’s tool-call format into this contract and preserves required call IDs. It also sends the tool schemas and complete relevant history to the provider. The harness itself owns permission checks, execution, and stopping.

Make the loop’s responsibilities visible

1. Initialize conversation history with the user’s task.
2. Ask the model adapter for one turn, passing available tool definitions.
3. If the turn is final, return it.
4. If it requests a tool, reject unknown names and validate the arguments.
5. Run the tool; append the call and its result as a matched pair.
6. Repeat until completion or a turn budget is reached.

Run a complete local example

This small program uses a fake model, so it runs with Python alone. The scripted model first asks the harness to count words; on its second turn it checks the tool observation and returns a final answer. The function and adapter share an explicit contract that you can later translate to your chosen model SDK.

from collections.abc import Callable
from typing import Any

Turn = dict[str, Any]
Tool = Callable[..., dict[str, int]]

def count_words(text: str) -> dict[str, int]:
    """Count whitespace-separated words in a text string."""
    return {"count": len(text.split())}

TOOLS: dict[str, Tool] = {"count_words": count_words}

def run_agent(
    task: str,
    ask_model: Callable[[list[dict[str, Any]], list[str]], Turn],
    max_turns: int = 4,
) -> str:
    history: list[dict[str, Any]] = [{"role": "user", "content": task}]

    for _ in range(max_turns):
        turn = ask_model(history, list(TOOLS))
        if turn.get("type") == "final":
            return turn["text"]
        if turn.get("type") != "tool_call":
            raise ValueError("Expected final or tool_call response")

        name = turn.get("name")
        arguments = turn.get("arguments")
        if name not in TOOLS:
            raise ValueError(f"Tool is not allowed: {name}")
        if not isinstance(arguments, dict):
            raise ValueError("Tool arguments must be an object")

        result = TOOLS[name](**arguments)
        history.extend([
            {"role": "assistant", "content": turn},
            {"role": "tool", "name": name,
             "content": {"ok": True, "data": result}},
        ])

    raise TimeoutError("Agent exceeded its turn limit")

# Deterministic test: no LLM, API key, or network required.
turns = iter([
    {"type": "tool_call", "name": "count_words",
     "arguments": {"text": "red green blue"}},
    {"type": "final", "text": "The phrase has 3 words."},
])

def fake_model(history, available_tools):
    assert "count_words" in available_tools
    if history[-1]["role"] == "tool":
        assert history[-1]["content"]["data"]["count"] == 3
    return next(turns)

assert run_agent("Count these words", fake_model) == \
    "The phrase has 3 words."

The assertion proves two things: the harness exposed the intended tool, and the tool’s observation reached the next model turn. It does not prove a real model will choose that tool correctly. That requires integration tests and task evaluations, covered in Lesson 9.

What this starter deliberately leaves out

It has no provider adapter, concurrent tool calls, persistent memory, network access, or approval UI. Add timeouts around real model and tool calls, map provider call IDs correctly, convert expected tool errors into safe observations, and authorize each side effect before expanding the capability set. Keep the orchestration itself small and testable.

Ask your teacher to adapt the normalized turn contract to the model API you plan to use.