Skip to content
Docs Pydantic Docs

Cancel and Resume

Cancel a streaming agent response from an interactive terminal, then continue the conversation with its preserved history.

Demonstrates:

Running the Example

With dependencies installed and environment variables set, run:

Terminal
python -m pydantic_ai_examples.cancel_and_resume

Press ++esc++ while a response is streaming to cancel that turn. The partial history is kept, so the next prompt continues the same conversation. Press ++ctrl+c++ or ++ctrl+d++ to exit.

How Cancellation and Resume Work

The reusable stream_turn function accepts a cancellation token and returns the history captured before cancellation. Resume with that history and a fresh token:

import asyncio
from collections.abc import AsyncIterator

from pydantic_ai_examples.cancel_and_resume import stream_turn

from pydantic_ai import Agent, CancellationToken, ModelMessage
from pydantic_ai.models.function import AgentInfo, FunctionModel

first_delta = asyncio.Event()


async def stream_response(
    _messages: list[ModelMessage], _info: AgentInfo
) -> AsyncIterator[str]:
    for chunk in ('You ', 'can ', 'resume ', 'this.'):
        yield chunk
        await asyncio.sleep(0.01)


async def cancel_on_first_delta(token: CancellationToken) -> None:
    await first_delta.wait()
    token.cancel()  # in the interactive example, the Esc key handler calls this


async def main() -> None:
    agent = Agent(FunctionModel(stream_function=stream_response))

    # First turn: cancel the run as soon as the response starts streaming.
    token = CancellationToken()
    canceller = asyncio.create_task(cancel_on_first_delta(token))
    history, cancelled = await stream_turn(
        agent, 'Tell me a story', [], token, lambda _text: first_delta.set()
    )
    await canceller
    print(f'cancelled={cancelled}, kept {len(history)} messages')
    #> cancelled=True, kept 2 messages

    # The next turn resumes from that preserved history with a fresh token.
    history, cancelled = await stream_turn(
        agent, 'Continue', history, CancellationToken(), lambda _text: None
    )
    print(f'cancelled={cancelled}, now {len(history)} messages')
    #> cancelled=False, now 4 messages


asyncio.run(main())

RunCancelled.all_messages() preserves completed work and partial streamed responses. Any interrupted tool calls are repaired automatically when the history is used by the next run.

Example Code

cancel_and_resume.py
from __future__ import annotations as _annotations

import asyncio
from collections.abc import Callable

import logfire
from prompt_toolkit import Application, PromptSession
from prompt_toolkit.formatted_text import FormattedText
from prompt_toolkit.key_binding import KeyBindings, KeyPressEvent
from prompt_toolkit.layout import Layout
from prompt_toolkit.layout.containers import Window
from prompt_toolkit.layout.controls import FormattedTextControl

from pydantic_ai import Agent, CancellationToken, ModelMessage, RunCancelled


async def stream_turn(
    agent: Agent[None],
    prompt: str,
    history: list[ModelMessage],
    token: CancellationToken,
    on_text: Callable[[str], None],
) -> tuple[list[ModelMessage], bool]:
    """Stream one turn, returning resumable history and whether it was cancelled."""
    try:
        async with agent.run_stream(
            prompt, message_history=history, cancellation_token=token
        ) as result:
            async for delta in result.stream_text(delta=True):
                on_text(delta)
        return result.all_messages(), False
    except RunCancelled as exc:
        return exc.all_messages(), True


async def _run_interactive_turn(
    agent: Agent[None],
    prompt: str,
    history: list[ModelMessage],
) -> list[ModelMessage]:
    token = CancellationToken()
    chunks: list[str] = []
    bindings = KeyBindings()

    def cancel_turn(_event: KeyPressEvent) -> None:
        token.cancel()

    # A custom `Application` doesn't get `PromptSession`'s Ctrl-C handling, so bind it here too.
    bindings.add('escape')(cancel_turn)
    bindings.add('c-c')(cancel_turn)

    control = FormattedTextControl(
        lambda: FormattedText([('class:answer', f'Agent: {"".join(chunks)}')])
    )
    application: Application[None] = Application(
        layout=Layout(Window(control)),
        key_bindings=bindings,
        full_screen=False,
    )

    def on_text(delta: str) -> None:
        chunks.append(delta)
        application.invalidate()

    async def run_turn() -> tuple[list[ModelMessage], bool]:
        try:
            return await stream_turn(agent, prompt, history, token, on_text)
        finally:
            application.exit()

    turn_task = asyncio.create_task(run_turn())
    await application.run_async()
    messages, was_cancelled = await turn_task
    if was_cancelled:
        print('⏹ cancelled')
    return messages


async def main() -> None:
    agent = Agent('openai:gpt-5-mini')
    session: PromptSession[str] = PromptSession()
    history: list[ModelMessage] = []

    print(
        'Chat with Pydantic AI. Press Esc or Ctrl-C to cancel a response; Ctrl-C or Ctrl-D at the prompt to exit.'
    )
    while True:
        try:
            prompt = await session.prompt_async('\nYou: ')
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if not prompt.strip():
            continue

        # A cancelled token stays cancelled, so each turn gets a fresh one.
        history = await _run_interactive_turn(agent, prompt, history)


if __name__ == '__main__':
    # 'if-token-present' means nothing will be sent (and the example will work) if you don't have logfire configured.
    # Configured here rather than at module level so importing `stream_turn` (e.g. from the docs) doesn't instrument.
    logfire.configure(send_to_logfire='if-token-present')
    logfire.instrument_pydantic_ai()
    asyncio.run(main())