---
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, by default, `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.  This
can be changed by overriding by the `LOOP_BUFFER` class variable in an
`Agent` subclass: with `None`, the default, an unbounded number of
events will be buffered. With `0`, the `loop` will run in lockstep.

## 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)