---
title: Custom Loops
description: Customize agent control flow.
type: guide
summary: Override the agent loop to customize scheduling, routing, history updates, logging, and tool execution.
---

# Custom Loops



Override `loop` when you need custom tool dispatch, logging, durability, or
branching. While the SDK defines the loop as a Python async generator, it still
uses a few framework components. Reuse those to retain framework-controlled
behavior.

Note that `loop` does not run in lock-step with the code that iterates the
stream. The framework runs it as a separate asyncio task that puts events on a
queue, and the `agent.run` stream reads from that queue. This means the loop
can keep making progress (for example, launching tools) while the consumer is
between reads, and events you `yield` may be consumed later than they are
produced.

## Explore the standard loop shape

The default loop keeps running while there is more work to do. On each
turn, it streams the model response and schedules tool calls:

```python
class CustomAgent(ai.Agent):
    # ai.Context keeps track of message history, run inputs, and other per-run data
    async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
        while context.keep_running():
            # call ai.stream with whatever model, tools, and messages the user has
            # passed to agent.run
            async with (
                ai.stream(context=context) as stream,
                # ai.ToolRunner is responsible for concurrent tool scheduling and
                # streaming, as well as graceful handling of hook interruptions
                ai.ToolRunner() as tool_runner,

            ):
                # ai.util.merge interleaves two streams together, so the agent
                # can fire off tools as they arrive in the stream rather than
                # waiting for the model to stop streaming; and also start streaming
                # the streaming tools.
                async for event in ai.util.merge(stream, tool_runner.events()):
                    yield event

                    if isinstance(event, ai.events.ToolEnd):
                        # context.resolve looks up the Python function for the
                        # tool name in the model's tool call
                        tool_call = context.resolve(event.tool_call)
                        # schedule the tool for concurrent execution
                        # using the tool runner
                        tool_runner.schedule(tool_call)

                # add new messages to message history stored in the context.
                # this works with internal replay machinery to enable serverless
                # execution.
                context.add(stream.message)
                context.add(tool_runner.get_tool_message())
```

You can modify most of the loop and extend `CustomAgent` to fit your application's needs,
and the SDK will keep working as long as the loop is still an async generator of events.


---

For a semantic overview of all documentation, see [/sitemap.md](/sitemap.md)

For an index of all available documentation, see [/llms.txt](/llms.txt)

For agent-facing discovery, including API and MCP surfaces, see [/agents.md](/agents.md)