Pydantic AI Agent: пошаговый контроль и отмена через iter и CancellationToken

Title:

URL Source: https://raw.githubusercontent.com/pydantic/pydantic-ai/main/docs/agent.md

Markdown Content: ## Introduction

Agents are Pydantic AI's primary interface for interacting with LLMs.

In some use cases a single Agent will control an entire application or component, but multiple agents can also interact to embody more complex workflows.

The [`Agent`][pydantic_ai.Agent] class has full API documentation, but conceptually you can think of an agent as a container for:

| **Component** | **Description** | | --------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- | | [Instructions](#instructions) | A set of instructions for the LLM written by the developer. | | [Function tool(s)](tools.md) and [toolsets](toolsets.md) | Functions that the LLM may call to get information while generating a response. | | [Structured output type](output.md) | The structured datatype the LLM must return at the end of a run, if specified. | | [Dependency type constraint](dependencies.md) | Dynamic instructions functions, tools, and output functions may all use dependencies when they're run. | | [LLM model](api/models/base.md) | Optional default LLM model associated with the agent. Can also be specified when running the agent. | | [Model Settings](#additional-configuration) | Optional default model settings to help fine tune requests. Can also be specified when running the agent. | | [Capabilities](capabilities/overview.md) | Reusable bundles of tools, hooks, instructions, and model settings that extend agent behavior. |

While each of these can be configured individually, [capabilities](capabilities/overview.md) let you bundle related behavior into reusable units that are easier to compose, share, and [load from configuration files](agent-spec.md).

In typing terms, agents are generic in their dependency and output types, e.g., an agent which required dependencies of type `#!python Foobar` and produced outputs of type `#!python list[str]` would have type `Agent[Foobar, list[str]]`. In practice, you shouldn't need to care about this, it should just mean your IDE can tell you when you have the right type, and if you choose to use [static type checking](#static-type-checking) it should work well with Pydantic AI.

Here's a toy example of an agent that simulates a roulette wheel:

```python {title="roulette_wheel.py"} from pydantic_ai import Agent, RunContext

roulette_agent = Agent( # (1)! 'openai:gpt-5.2', deps_type=int, output_type=bool, system_prompt=( 'Use the `roulette_wheel` function to see if the ' 'customer has won based on the number they provide.' ), )

@roulette_agent.tool async def roulette_wheel(ctx: RunContext[int], square: int) -> str: # (2)! """check if the square is a winner""" return 'winner' if square == ctx.deps else 'loser'

# Run the agent success_number = 18 # (3)! result = roulette_agent.run_sync('Put my money on square eighteen', deps=success_number) print(result.output) # (4)! #> True

result = roulette_agent.run_sync('I bet five is the winner', deps=success_number) print(result.output) #> False ```

1. Create an agent, which expects an integer dependency and produces a boolean output. This agent will have type `#!python Agent[int, bool]`. 2. Define a tool that checks if the square is a winner. Here [`RunContext`][pydantic_ai.tools.RunContext] is parameterized with the dependency type `int`; if you got the dependency type wrong you'd get a typing error. 3. In reality, you might want to use a random number here e.g. `random.randint(0, 36)`. 4. `result.output` will be a boolean indicating if the square is a winner. Pydantic performs the output validation, and it'll be typed as a `bool` since its type is derived from the `output_type` generic parameter of the agent.

!!! tip "Agents are designed for reuse, like FastAPI Apps" You can instantiate one agent and use it globally throughout your application, as you would a small [FastAPI][fastapi.FastAPI] app or an [APIRouter][fastapi.APIRouter], or dynamically create as many agents as you want. Both are valid and supported ways to use agents.

## Running Agents

There are five ways to run an agent:

1. [`agent.run()`][pydantic_ai.agent.AbstractAgent.run] — an async function which returns a [`RunResult`][pydantic_ai.agent.AgentRunResult] containing a completed response. 2. [`agent.run_sync()`][pydantic_ai.agent.AbstractAgent.run_sync] — a plain, synchronous function which returns a [`RunResult`][pydantic_ai.agent.AgentRunResult] containing a completed response (internally, this just calls `loop.run_until_complete(self.run())`). 3. [`agent.run_stream()`][pydantic_ai.agent.AbstractAgent.run_stream] — an async context manager which returns a [`StreamedRunResult`][pydantic_ai.result.StreamedRunResult], which contains methods to stream text and structured output as an async iterable. [`agent.run_stream_sync()`][pydantic_ai.agent.AbstractAgent.run_stream_sync] is a synchronous variation that returns a [`StreamedRunResultSync`][pydantic_ai.result.StreamedRunResultSync] with synchronous versions of the same methods. 4. [`agent.run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events] — an async context manager which yields an async iterator over [`AgentStreamEvent`s][pydantic_ai.messages.AgentStreamEvent] ending with an [`AgentRunResultEvent`][pydantic_ai.run.AgentRunResultEvent] containing the final run result. 5. [`agent.iter()`][pydantic_ai.agent.Agent.iter] — a context manager which returns an [`AgentRun`][pydantic_ai.agent.AgentRun], an async iterable over the nodes of the agent's underlying [`Graph`][pydantic_graph.graph_builder.Graph].

Here's a simple example demonstrating the first four:

```python {title="run_agent.py"} from pydantic_ai import Agent, AgentRunResultEvent, AgentStreamEvent

agent = Agent('openai:gpt-5.2')

result_sync = agent.run_sync('What is the capital of Italy?') print(result_sync.output) #> The capital of Italy is Rome.

async def main(): result = await agent.run('What is the capital of France?') print(result.output) #> The capital of France is Paris.

async with agent.run_stream('What is the capital of the UK?') as response: async for text in response.stream_text(): print(text) #> The capital of #> The capital of the UK is #> The capital of the UK is London.

collected: list[AgentStreamEvent | AgentRunResultEvent] = [] async with agent.run_stream_events('What is the capital of Mexico?') as events: async for event in events: collected.append(event) print(collected) """ [ PartStartEvent(index=0, part=TextPart(content='The capital of ')), FinalResultEvent(tool_name=None, tool_call_id=None), PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='Mexico is Mexico ')), PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='City.')), PartEndEvent( index=0, part=TextPart(content='The capital of Mexico is Mexico City.') ), AgentRunResultEvent( result=AgentRunResult(output='The capital of Mexico is Mexico City.') ), ] """ ```

_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

You can also pass messages from previous runs to continue a conversation or provide context, as described in [Messages and Chat History](message-history.md).

### Streaming Events and Final Output

As shown in the example above, [`run_stream()`][pydantic_ai.agent.AbstractAgent.run_stream] makes it easy to stream the agent's final output as it comes in. It also takes an optional `event_stream_handler` argument that you can use to gain insight into what is happening during the run before the final output is produced. During a realtime session, the same handler stream can also contain realtime-only [`RealtimeEvent`][pydantic_ai.realtime.RealtimeEvent] members.

The example below shows how to stream events and text output. You can also [stream structured output](output.md#streaming-structured-output).

!!! note The `run_stream()` and `run_stream_sync()` methods will consider the first output that matches the [output type](output.md#structured-output) (which could be text, an [output tool](output.md#tool-output) call, or a [deferred](deferred-tools.md) tool call) to be the final output of the agent run, even when the model generates (additional) tool calls after this "final" output.

These "dangling" tool calls will not be executed unless the agent's [`end_strategy`][pydantic_ai.agent.Agent.end_strategy] is set to `'graceful'` or `'exhaustive'`, and even then their results will not be sent back to the model as the agent run will already be considered completed. In short, if the model returns both tool calls and text, and the agent's output type is `str`, **the tool calls will not run** in streaming mode with the default setting.

If you want to always keep running the agent when it performs tool calls, and stream all events from the model's streaming response and the agent's execution of tools, use [`agent.run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events] or [`agent.iter()`][pydantic_ai.agent.AbstractAgent.iter] instead, as described in the following sections.

```python {title="run_stream_event_stream_handler.py"} import asyncio from collections.abc import AsyncIterable from datetime import date

from pydantic_ai import ( Agent, AgentStreamEvent, FinalResultEvent, FunctionToolCallEvent, FunctionToolResultEvent, PartDeltaEvent, PartStartEvent, RunContext, TextPartDelta, ThinkingPartDelta, ToolCallPartDelta, )

weather_agent = Agent( 'openai:gpt-5.2', system_prompt='Providing a weather forecast at the locations the user provides.', )

@weather_agent.tool async def weather_forecast( ctx: RunContext, location: str, forecast_date: date, ) -> str: return f'The forecast in {location} on {forecast_date} is 24°C and sunny.'

output_messages: list[str] = []

async def handle_event(event: AgentStreamEvent): if isinstance(event, PartStartEvent): output_messages.append(f'[Request] Starting part {event.index}: {event.part!r}') elif isinstance(event, PartDeltaEvent): if isinstance(event.delta, TextPartDelta): output_messages.append(f'[Request] Part {event.index} text delta: {event.delta.content_delta!r}') elif isinstance(event.delta, ThinkingPartDelta): output_messages.append(f'[Request] Part {event.index} thinking delta: {event.delta.content_delta!r}') elif isinstance(event.delta, ToolCallPartDelta): output_messages.append(f'[Request] Part {event.index} args delta: {event.delta.args_delta}') elif isinstance(event, FunctionToolCallEvent): output_messages.append( f'[Tools] The LLM calls tool={event.part.tool_name!r} with args={event.part.args} (tool_call_id={event.part.tool_call_id!r})' ) elif isinstance(event, FunctionToolResultEvent): output_messages.append(f'[Tools] Tool call {event.tool_call_id!r} returned => {event.part.content}') elif isinstance(event, FinalResultEvent): output_messages.append(f'[Result] The model starting producing a final result (tool_name={event.tool_name})')

async def event_stream_handler( ctx: RunContext, event_stream: AsyncIterable[AgentStreamEvent], ): async for event in event_stream: await handle_event(event)

async def main(): user_prompt = 'What will the weather be like in Paris on Tuesday?'

async with weather_agent.run_stream(user_prompt, event_stream_handler=event_stream_handler) as run: async for output in run.stream_text(): output_messages.append(f'[Output] {output}')

if __name__ == '__main__': asyncio.run(main())

print(output_messages) """ [ "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')", '[Request] Part 0 args delta: {"location":"Pa', '[Request] Part 0 args delta: ris","forecast_', '[Request] Part 0 args delta: date":"2030-01-', '[Request] Part 0 args delta: 01"}', '[Tools] The LLM calls tool=\'weather_forecast\' with args={"location":"Paris","forecast_date":"2030-01-01"} (tool_call_id=\'0001\')', "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.", "[Request] Starting part 0: TextPart(content='It will be ')", '[Result] The model starting producing a final result (tool_name=None)', '[Output] It will be ', '[Output] It will be warm and sunny ', '[Output] It will be warm and sunny in Paris on ', '[Output] It will be warm and sunny in Paris on Tuesday.', ] """ ```

_(This example is complete, it can be run "as is")_

### Streaming All Events

Like `agent.run_stream()`, [`agent.run()`][pydantic_ai.agent.AbstractAgent.run_stream] takes an optional `event_stream_handler` argument that lets you stream all events from the model's streaming response and the agent's execution of tools. Unlike `run_stream()`, it always runs the agent graph to completion even if text was received ahead of tool calls that looked like it could've been the final result. During a realtime session, an event stream handler can also receive realtime-only [`RealtimeEvent`][pydantic_ai.realtime.RealtimeEvent] members.

For convenience, a [`agent.run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events] method is also available as a wrapper around `run(event_stream_handler=...)`. It is an async context manager that yields an async iterator over [`AgentStreamEvent`s][pydantic_ai.messages.AgentStreamEvent] ending with an [`AgentRunResultEvent`][pydantic_ai.run.AgentRunResultEvent] carrying the final run result.

!!! note As they return raw events as they come in, the `run_stream_events()` and `run(event_stream_handler=...)` methods require you to piece together the streamed text and structured output yourself from the `PartStartEvent` and subsequent `PartDeltaEvent`s.

To get the best of both worlds, at the expense of some additional complexity, you can use [`agent.iter()`][pydantic_ai.agent.AbstractAgent.iter] as described in the next section, which lets you [iterate over the agent graph](#iterating-over-an-agents-graph) and [stream both events and output](#streaming-all-events-and-output) at every step. See [Making structured responses appear faster](output.md#making-structured-responses-appear-faster) for a focused example using validated structured output.

```python {title="run_events.py" requires="run_stream_event_stream_handler.py"} import asyncio

from pydantic_ai import AgentRunResultEvent

from run_stream_event_stream_handler import handle_event, output_messages, weather_agent

async def main(): user_prompt = 'What will the weather be like in Paris on Tuesday?'

async with weather_agent.run_stream_events(user_prompt) as events: async for event in events: if isinstance(event, AgentRunResultEvent): output_messages.append(f'[Final Output] {event.result.output}') else: await handle_event(event)

if __name__ == '__main__': asyncio.run(main())

print(output_messages) """ [ "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')", '[Request] Part 0 args delta: {"location":"Pa', '[Request] Part 0 args delta: ris","forecast_', '[Request] Part 0 args delta: date":"2030-01-', '[Request] Part 0 args delta: 01"}', '[Tools] The LLM calls tool=\'weather_forecast\' with args={"location":"Paris","forecast_date":"2030-01-01"} (tool_call_id=\'0001\')', "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.", "[Request] Starting part 0: TextPart(content='It will be ')", '[Result] The model starting producing a final result (tool_name=None)', "[Request] Part 0 text delta: 'warm and sunny '", "[Request] Part 0 text delta: 'in Paris on '", "[Request] Part 0 text delta: 'Tuesday.'", '[Final Output] It will be warm and sunny in Paris on Tuesday.', ] """ ```

_(This example is complete, it can be run "as is")_

### Iterating Over an Agent's Graph

Under the hood, each `Agent` in Pydantic AI uses **pydantic-graph** to manage its execution flow. **pydantic-graph** is a generic, type-centric library for building and running finite state machines in Python. It doesn't actually depend on Pydantic AI — you can use it standalone for workflows that have nothing to do with GenAI — but Pydantic AI makes use of it to orchestrate the handling of model requests and model responses in an agent's run.

In many scenarios, you don't need to worry about pydantic-graph at all; calling `agent.run(...)` simply traverses the underlying graph from start to finish. However, if you need deeper insight or control — for example to inject your own logic at specific stages — Pydantic AI exposes the lower-level iteration process via [`Agent.iter`][pydantic_ai.agent.Agent.iter]. This method returns an [`AgentRun`][pydantic_ai.agent.AgentRun], which you can async-iterate over, or manually drive node-by-node via the [`next`][pydantic_ai.agent.AgentRun.next] method. Once the agent's graph returns an [`End`][pydantic_graph.basenode.End], you have the final result along with a detailed history of all steps.

#### `async for` iteration

Here's an example of using `async for` with `iter` to record each node the agent executes:

```python {title="agent_iter_async_for.py"} from pydantic_ai import Agent

agent = Agent('openai:gpt-5.2')

async def main(): nodes = [] # Begin an AgentRun, which is an async-iterable over the nodes of the agent's graph async with agent.iter('What is the capital of France?') as agent_run: async for node in agent_run: # Each node represents a step in the agent's execution nodes.append(node) print(nodes) """ [ UserPromptNode( user_prompt='What is the capital of France?', instructions_functions=[], system_prompts=(), system_prompt_functions=[], system_prompt_dynamic_functions={}, ), ModelRequestNode( request=ModelRequest( parts=[ UserPromptPart( content='What is the capital of France?', timestamp=datetime.datetime(...), ) ], timestamp=datetime.datetime(...), run_id='...', conversation_id='...', ) ), CallToolsNode( model_response=ModelResponse( parts=[TextPart(content='The capital of France is Paris.')], usage=RequestUsage( cost=Decimal('0.000196'), input_tokens=56, output_tokens=7 ), model_name='gpt-5.2', timestamp=datetime.datetime(...), run_id='...', conversation_id='...', ) ), End(data=FinalResult(output='The capital of France is Paris.')), ] """ print(agent_run.result.output) #> The capital of France is Paris. ```

_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

- The `AgentRun` is an async iterator that yields each node (`BaseNode` or `End`) in the flow. - The run ends when an `End` node is returned.

#### Using `.next(...)` manually

You can also drive the iteration manually by passing the node you want to run next to the `AgentRun.next(...)` method. This allows you to inspect or modify the node before it executes or skip nodes based on your own logic, and to catch errors in `next()` more easily:

```python {title="agent_iter_next.py"} from pydantic_ai import Agent from pydantic_graph import End

agent = Agent('openai:gpt-5.2')

async def main(): async with agent.iter('What is the capital of France?') as agent_run: node = agent_run.next_node # (1)!

all_nodes = [node]

# Drive the iteration manually: while not isinstance(node, End): # (2)! node = await agent_run.next(node) # (3)! all_nodes.append(node) # (4)!

print(all_nodes) """ [ UserPromptNode( user_prompt='What is the capital of France?', instructions_functions=[], system_prompts=(), system_prompt_functions=[], system_prompt_dynamic_functions={}, ), ModelRequestNode( request=ModelRequest( parts=[ UserPromptPart( content='What is the capital of France?', timestamp=datetime.datetime(...), ) ], timestamp=datetime.datetime(...), run_id='...', conversation_id='...', ) ), CallToolsNode( model_response=ModelResponse( parts=[TextPart(content='The capital of France is Paris.')], usage=RequestUsage( cost=Decimal('0.000196'), input_tokens=56, output_tokens=7 ), model_name='gpt-5.2', timestamp=datetime.datetime(...), run_id='...', conversation_id='...', ) ), End(data=FinalResult(output='The capital of France is Paris.')), ] """ ```

1. We start by grabbing the first node that will be run in the agent's graph. 2. The agent run is finished once an `End` node has been produced; instances of `End` cannot be passed to `next`. 3. When you call `await agent_run.next(node)`, it executes that node in the agent's graph, updates the run's history, and returns the _next_ node to run. 4. You could also inspect or mutate the new `node` here as needed.

_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

#### Accessing usage and final output

You can retrieve usage statistics (tokens, requests, etc.) at any time from the [`AgentRun`][pydantic_ai.agent.AgentRun] object via `agent_run.usage`. This property returns a [`RunUsage`][pydantic_ai.usage.RunUsage] object containing the usage data.

[`RunUsage.cost`][pydantic_ai.usage.RunUsage.cost] additionally holds a best-effort estimate of the run's total cost in USD, calculated from each request's usage with [genai-prices](https://github.com/pydantic/genai-prices). Requests to models or providers that genai-prices doesn't have pricing data for don't contribute to the total.

Once the run finishes, `agent_run.result` becomes an [`AgentRunResult`][pydantic_ai.agent.AgentRunResult] object containing the final output (and related metadata).

#### Streaming All Events and Output

Here is an example of streaming an agent run in combination with `async for` iteration:

```python {title="streaming_iter.py"} import asyncio from dataclasses import dataclass from datetime import date

from pydantic_ai import ( Agent, FinalResultEvent, FunctionToolCallEvent, FunctionToolResultEvent, PartDeltaEvent, PartStartEvent, RunContext, TextPartDelta, ThinkingPartDelta, ToolCallPartDelta, )

@dataclass class WeatherService: async def get_forecast(self, location: str, forecast_date: date) -> str: # In real code: call weather API, DB queries, etc. return f'The forecast in {location} on {forecast_date} is 24°C and sunny.'

async def get_historic_weather(self, location: str, forecast_date: date) -> str: # In real code: call a historical weather API or DB return f'The weather in {location} on {forecast_date} was 18°C and partly cloudy.'

weather_agent = Agent[WeatherService, str]( 'openai:gpt-5.2', deps_type=WeatherService, output_type=str, # We'll produce a final answer as plain text system_prompt='Providing a weather forecast at the locations the user provides.', )

@weather_agent.tool async def weather_forecast( ctx: RunContext[WeatherService], location: str, forecast_date: date, ) -> str: if forecast_date >= date.today(): return await ctx.deps.get_forecast(location, forecast_date) else: return await ctx.deps.get_historic_weather(location, forecast_date)

output_messages: list[str] = []

async def main(): user_prompt = 'What will the weather be like in Paris on Tuesday?'

# Begin a node-by-node, streaming iteration async with weather_agent.iter(user_prompt, deps=WeatherService()) as run: async for node in run: if Agent.is_user_prompt_node(node): # A user prompt node => The user has provided input output_messages.append(f'=== UserPromptNode: {node.user_prompt} ===') elif Agent.is_model_request_node(node): # A model request node => We can stream tokens from the model's request output_messages.append('=== ModelRequestNode: streaming partial request tokens ===') async with node.stream(run.ctx) as request_stream: final_result_found = False async for event in request_stream: if isinstance(event, PartStartEvent): output_messages.append(f'[Request] Starting part {event.index}: {event.part!r}') elif isinstance(event, PartDeltaEvent): if isinstance(event.delta, TextPartDelta): output_messages.append( f'[Request] Part {event.index} text delta: {event.delta.content_delta!r}' ) elif isinstance(event.delta, ThinkingPartDelta): output_messages.append( f'[Request] Part {event.index} thinking delta: {event.delta.content_delta!r}' ) elif isinstance(event.delta, ToolCallPartDelta): output_messages.append( f'[Request] Part {event.index} args delta: {event.delta.args_delta}' ) elif isinstance(event, FinalResultEvent): output_messages.append( f'[Result] The model started producing a final result (tool_name={event.tool_name})' ) final_result_found = True break

if final_result_found: # Once the final result is found, we can call `AgentStream.stream_text()` to stream the text. # A similar `AgentStream.stream_output()` method is available to stream structured output. async for output in request_stream.stream_text(): output_messages.append(f'[Output] {output}') elif Agent.is_call_tools_node(node): # A handle-response node => The model returned some data, potentially calls a tool output_messages.append('=== CallToolsNode: streaming partial response & tool usage ===') async with node.stream(run.ctx) as handle_stream: async for event in handle_stream: if isinstance(event, FunctionToolCallEvent): output_messages.append( f'[Tools] The LLM calls tool={event.part.tool_name!r} with args={event.part.args} (tool_call_id={event.part.tool_call_id!r})' ) elif isinstance(event, FunctionToolResultEvent): output_messages.append( f'[Tools] Tool call {event.tool_call_id!r} returned => {event.part.content}' ) elif Agent.is_end_node(node): # Once an End node is reached, the agent run is complete assert run.result is not None assert run.result.output == node.data.output output_messages.append(f'=== Final Agent Output: {run.result.output} ===')

if __name__ == '__main__': asyncio.run(main())

print(output_messages) """ [ '=== UserPromptNode: What will the weather be like in Paris on Tuesday? ===', '=== ModelRequestNode: streaming partial request tokens ===', "[Request] Starting part 0: ToolCallPart(tool_name='weather_forecast', tool_call_id='0001')", '[Request] Part 0 args delta: {"location":"Pa', '[Request] Part 0 args delta: ris","forecast_', '[Request] Part 0 args delta: date":"2030-01-', '[Request] Part 0 args delta: 01"}', '=== CallToolsNode: streaming partial response & tool usage ===', '[Tools] The LLM calls tool=\'weather_forecast\' with args={"location":"Paris","forecast_date":"2030-01-01"} (tool_call_id=\'0001\')', "[Tools] Tool call '0001' returned => The forecast in Paris on 2030-01-01 is 24°C and sunny.", '=== ModelRequestNode: streaming partial request tokens ===', "[Request] Starting part 0: TextPart(content='It will be ')", '[Result] The model started producing a final result (tool_name=None)', '[Output] It will be ', '[Output] It will be warm and sunny ', '[Output] It will be warm and sunny in Paris on ', '[Output] It will be warm and sunny in Paris on Tuesday.', '=== CallToolsNode: streaming partial response & tool usage ===', '=== Final Agent Output: It will be warm and sunny in Paris on Tuesday. ===', ] """ ```

_(This example is complete, it can be run "as is")_

### Cancelling a Run

A run in flight can be cancelled entirely -- e.g. when a user hits a "stop" button. Create a [`CancellationToken`][pydantic_ai.CancellationToken], pass it to the run, and call `cancel()` from the stop handler. Cancellation raises [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] with the completed message history and usage so you can persist and resume the conversation:

```python {title="run_cancel.py"} import asyncio

from pydantic_ai import Agent, CancellationToken, RunCancelled

agent = Agent('test') tool_started = asyncio.Event()

@agent.tool_plain async def slow_lookup() -> str: tool_started.set() await asyncio.sleep(10) return 'result'

async def main(): token = CancellationToken() run = asyncio.create_task( agent.run('Look something up', cancellation_token=token) ) await tool_started.wait() token.cancel() # (1)!

try: await run except RunCancelled as exc: messages = exc.all_messages() print(f'Cancelled after {len(messages)} messages') #> Cancelled after 2 messages await agent.run(message_history=messages) # (2)! ```

1. `cancel()` is idempotent and thread-safe. One token may govern multiple concurrent runs, cancelling all of them. A token is single-use: once cancelled it stays cancelled, and passing an already-cancelled token to a run prevents that run from starting (which also closes the "cancel raced ahead of the run" gap). So mint a fresh token per run or per stop gesture -- reusing one token across a session would cancel every run after the first before it starts. 2. [`RunCancelled.all_messages()`][pydantic_ai.exceptions.RunCancelled.all_messages] contains everything completed before cancellation, including completed tool results. Any dangling tool call is [repaired automatically](message-history.md#making-histories-provider-valid) when the history is resumed.

[UI adapter](ui/overview.md) users can persist this resumable history with the `on_cancel` callback.

_(This example is complete, it can be run "as is" -- you'll need to add `asyncio.run(main())` to run `main`)_

[`agent.run_sync()`][pydantic_ai.agent.AbstractAgent.run_sync] accepts the same token. Calling `token.cancel()` from another thread is the only way to interrupt a synchronous run while it is blocked.

!!! note "Which mechanism, and which exception" A [`CancellationToken`][pydantic_ai.CancellationToken] is the one to reach for by default -- it's the only surface that works from outside the run, from another thread, and against `run_sync()`, and one token can govern several runs at once. The others exist for where a token can't reach:

| Where you are when you cancel | Use | Run ends with | | --- | --- | --- | | Outside the run (a "stop" button, another thread) | [`CancellationToken`][pydantic_ai.CancellationToken] | [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] | | Inside a tool, `event_stream_handler`, or capability hook | [`RunContext.cancel()`][pydantic_ai.tools.RunContext.cancel] | [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] | | Consuming [`run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events] | [`AgentRunEvents.cancel()`][pydantic_ai.agent.AgentRunEvents.cancel] on the yielded handle | [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] | | Driving the graph yourself via [`agent.iter()`][pydantic_ai.agent.Agent.iter] | [`AgentRun.cancel()`][pydantic_ai.run.AgentRun.cancel] | [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] | | The environment cancelled you (`asyncio.timeout()`, a [`TaskGroup`][asyncio.TaskGroup], shutdown) | *(you don't call anything)* | [`CancelledError`][asyncio.CancelledError] |

The first four are **first-party**: Pydantic AI stops the run itself and raises `RunCancelled`, an ordinary catchable exception carrying the resumable history. The last is **external**: the `CancelledError` keeps propagating unchanged -- so `asyncio.timeout()` still raises `TimeoutError`, a `TaskGroup` still tears down, and Temporal still ends the workflow *Cancelled* -- with the same history *attached* for [`RunCancelled.from_cancellation()`][pydantic_ai.exceptions.RunCancelled.from_cancellation]. Pydantic AI can't turn an external `CancelledError` into `RunCancelled` without breaking those semantics; that's why cancellation has the two shapes, covered next.

When the surrounding environment cancels the run -- for example through `asyncio.timeout()`, a [`TaskGroup`][asyncio.TaskGroup], or application shutdown -- the [`CancelledError`][asyncio.CancelledError] remains unchanged. [`RunCancelled.from_cancellation()`][pydantic_ai.exceptions.RunCancelled.from_cancellation] provides the attached run state:

```python {title="run_external_cancel.py"} import asyncio

from pydantic_ai import Agent, RunCancelled

agent = Agent('test') tool_started = asyncio.Event()

@agent.tool_plain async def slow_lookup() -> str: tool_started.set() await asyncio.sleep(10) return 'result'

async def main(): task = asyncio.create_task(agent.run('Look something up')) await tool_started.wait() task.cancel() # (1)!

try: await task except asyncio.CancelledError as exc: cancelled = RunCancelled.from_cancellation(exc) # (2)! assert cancelled is not None messages = cancelled.all_messages() print(f'Cancelled after {len(messages)} messages') #> Cancelled after 2 messages await agent.run(message_history=messages) # (3)! ```

1. This demonstrates cancellation imposed by the surrounding asyncio environment. For application stop gestures, prefer a `CancellationToken`. 2. External cancellation is never converted: `asyncio.timeout()`, [`TaskGroup`][asyncio.TaskGroup], and [Temporal](durable_execution/temporal.md) cancellation semantics are preserved. The run state rides along on the original `CancelledError`. 3. [`RunCancelled.all_messages()`][pydantic_ai.exceptions.RunCancelled.all_messages] contains everything completed before cancellation, including completed tool results. Any dangling tool call is [repaired automatically](message-history.md#making-histories-provider-valid) when the history is resumed.

_(This example is complete, it can be run "as is" -- you'll need to add `asyncio.run(main())` to run `main`)_

On Python 3.10, asyncio recreates `CancelledError` across an `await task` boundary, but chains the original exception -- carrying the attached run state -- via `__context__`, which `from_cancellation()` traverses. The chain is attached only to the first `await` of the cancelled task, so later awaits of the same task see an unchained exception; [`capture_run_messages()`][pydantic_ai.agent.capture_run_messages] is the fallback when only history is needed.

When consuming [`run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events], the yielded [`AgentRunEvents`][pydantic_ai.agent.AgentRunEvents] handle offers a first-party alternative that needs no task juggling: [`AgentRunEvents.cancel()`][pydantic_ai.agent.AgentRunEvents.cancel] is safe to call from another task (e.g. a UI's "stop" handler) and surfaces as `RunCancelled` on continued iteration:

```python {title="run_cancel_stream_events.py"} from pydantic_ai import Agent, RunCancelled

agent = Agent('test')

async def main(): async with agent.run_stream_events('Write a long essay about Python') as events: try: async for _event in events: events.cancel() # (1)! except RunCancelled as exc: print(f'Cancelled after {len(exc.all_messages())} messages') #> Cancelled after 2 messages ```

1. Idempotent, a no-op once the run has finished, and callable before the first iteration to prevent the run from starting at all.

_(This example is complete, it can be run "as is" -- you'll need to add `asyncio.run(main())` to run `main`)_

Externally cancelling the consuming task works here too: the background run tears down, the propagating `CancelledError` carries the run state for `from_cancellation()`, and the handle's `all_messages()` and `usage` remain accessible afterwards.

To request cancellation from a tool, an `event_stream_handler`, or a capability hook, call [`RunContext.cancel()`][pydantic_ai.tools.RunContext.cancel]. This requests first-party cancellation, so the run ends with [`RunCancelled`][pydantic_ai.exceptions.RunCancelled] rather than an external `CancelledError`. `cancel()` itself returns normally — the cancellation is delivered at the calling code's next `await`, and the tool's return value is discarded — so a tool can still run cleanup after requesting it:

```python {title="run_cancel_from_tool.py"} from pydantic_ai import Agent, RunCancelled, RunContext

agent = Agent('test')

@agent.tool async def stop(ctx: RunContext) -> str: ctx.cancel() return 'discarded' # cancel() returned; this value is never sent to the model

async def main(): try: await agent.run('Stop now') except RunCancelled as exc: print(f'Cancelled a