ai-python-custom-loop

Keep the default shape unless you must change control flow:

class MyAgent(ai.Agent):
    async def loop(self, context: ai.Context):
        while context.keep_running():
            async with (
                ai.stream(context=context) as stream,
                ai.ToolRunner() as runner,
            ):
                async for event in ai.util.merge(stream, runner.events()):
                    yield event

                    if isinstance(event, ai.events.ToolEnd):
                        runner.schedule(context.resolve(event.tool_call))

                context.add(stream.message)
                context.add(runner.get_tool_message())

Rules:

• Call context.keep_running() at the top of each turn.
• Use ai.stream(context=context) so model, messages, tools, output type, and params stay together.
• Yield events from the loop. Agent.run hides replay events from callers.
• On ToolEnd, use context.resolve(event.tool_call). It handles validation, approval gates, and cached replay results.
• Do not call tool.fn directly unless you also handle validation, approvals, and cached results.
• Schedule resolved calls with ToolRunner.schedule(...).
• ToolRunner.schedule(...) also accepts a zero-arg async callable that returns ai.events.ToolCallResult.
• If you make a result yourself, use runner.add_result(ai.tool_result(...)).
• Add stream.message, then runner.get_tool_message(). context.add(...) skips replay messages.
• Every tool call must get one tool result.
• For hooks, let context.resolve(...) build the gated call. Use ai-python-serverless-execution for request boundaries.
• For durable calls, keep this shape and wrap only model or tool I/O. Use ai-python-durable-execution.