Progress from a Pydantic AI agent, and a board to watch it on

Step through the run with agent.iter

agent.iter() returns an AgentRun you iterate node by node. Each node is one step: UserPromptNode, ModelRequestNode, CallToolsNode, or End. Stream a node to see what happens inside it: a CallToolsNode yields FunctionToolCallEvent and FunctionToolResultEvent, a ModelRequestNode yields PartStartEvent, PartDeltaEvent and FinalResultEvent.

from pydantic_ai import Agent
from pydantic_ai.messages import FunctionToolCallEvent, FunctionToolResultEvent

async with agent.iter("Upgrade the payments service to the new API") as run:
    async for node in run:
        if Agent.is_call_tools_node(node):
            async with node.stream(run.ctx) as events:
                async for event in events:
                    if isinstance(event, FunctionToolCallEvent):
                        print(f"> calling {event.part.tool_name}")
                    elif isinstance(event, FunctionToolResultEvent):
                        print("> tool finished")
        elif Agent.is_end_node(node):
            print("done:", run.result.output)

For a flat stream instead of nodes, run_stream_events() yields every AgentStreamEvent and ends with an AgentRunResultEvent holding the result, or pass event_stream_handler to run(). Either way you assemble streamed text yourself from PartStartEvent and PartDeltaEvent.

run_stream() can skip your tools

run_stream() treats the first output that matches the output type as final. With the default settings, if the model returns tool calls and text together and the output type is str, the tool calls don't run. For an agent whose work happens in tools, use agent.iter() or run_stream_events(), or set end_strategy to 'graceful' or 'exhaustive'.

Seeing a run from outside the process

Node streams and event handlers only reach the process that started the run. An agent in a worker, a container or a scheduled job has nobody watching it, and no way to ask a question and wait for the answer. It has to report somewhere you can see, which is what the live board is for.

Connect Taskhold's MCP server

Pydantic AI connects to remote MCP servers with MCPToolset from pydantic_ai.mcp; a URL not ending in /sse uses Streamable HTTP. An unattended agent can't do a browser sign-in, so it uses an API key: start the 7-day trial, open Settings > API keys and make one for this agent, and pass it as a header.

import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset

taskhold = MCPToolset(
    "https://taskhold.com/api/mcp",
    headers={"Authorization": f"Bearer {os.environ['TASKHOLD_API_KEY']}"},
)
agent = Agent("openai:gpt-6.1-sol", toolsets=[taskhold], instructions=WORKFLOW)

async with agent:  # opens and closes the MCP connection around the run
    result = await agent.run("Upgrade the payments service to the new API")

• Each run shows on the board as a task with a bar split by step, the agent's label, how long it has run and what it's doing now.
• A question the agent blocks on moves to the top of the board. It waits in taskhold_updates (up to 50 s per call) while you answer from any browser or your phone. If you set a custom http_client with a timeout, keep it above that.
• Logfire (logfire.instrument_pydantic_ai()) gives you traces of the same run for debugging; the board is for following the work and answering it.