`pydantic_ai.ag_ui`

为 Pydantic AI 代理提供一个 AG-UI 协议适配器。

此包提供了 pydantic-ai 代理与 ag-ui 之间的无缝集成，用于构建具有流式事件通信的交互式 AI 应用程序。

SSE_CONTENT_TYPE `模块属性`

SSE_CONTENT_TYPE: Final[str] = 'text/event-stream'

服务器发送事件 (SSE) 的内容类型标头值。

AGUIApp

基类：Generic[AgentDepsT, OutputDataT], Starlette

用于运行支持 AG-UI 协议的 Pydantic AI 代理的 ASGI 应用程序。

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

class AGUIApp(Generic[AgentDepsT, OutputDataT], Starlette):
    """ASGI application for running Pydantic AI agents with AG-UI protocol support."""

    def __init__(
        self,
        agent: AbstractAgent[AgentDepsT, OutputDataT],
        *,
        # Agent.iter parameters.
        output_type: OutputSpec[Any] | None = None,
        model: Model | KnownModelName | str | None = None,
        deps: AgentDepsT = None,
        model_settings: ModelSettings | None = None,
        usage_limits: UsageLimits | None = None,
        usage: RunUsage | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        # Starlette parameters.
        debug: bool = False,
        routes: Sequence[BaseRoute] | None = None,
        middleware: Sequence[Middleware] | None = None,
        exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
        on_startup: Sequence[Callable[[], Any]] | None = None,
        on_shutdown: Sequence[Callable[[], Any]] | None = None,
        lifespan: Lifespan[AGUIApp[AgentDepsT, OutputDataT]] | None = None,
    ) -> None:
        """An ASGI application that handles every AG-UI request by running the agent.

        Note that the `deps` will be the same for each request, with the exception of the AG-UI state that's
        injected into the `state` field of a `deps` object that implements the [`StateHandler`][pydantic_ai.ag_ui.StateHandler] protocol.
        To provide different `deps` for each request (e.g. based on the authenticated user),
        use [`pydantic_ai.ag_ui.run_ag_ui`][pydantic_ai.ag_ui.run_ag_ui] or
        [`pydantic_ai.ag_ui.handle_ag_ui_request`][pydantic_ai.ag_ui.handle_ag_ui_request] instead.

        Args:
            agent: The agent to run.

            output_type: Custom output type to use for this run, `output_type` may only be used if the agent has
                no output validators since output validators would expect an argument that matches the agent's
                output type.
            model: Optional model to use for this run, required if `model` was not set when creating the agent.
            deps: Optional dependencies to use for this run.
            model_settings: Optional settings to use for this model's request.
            usage_limits: Optional limits on model request count or token usage.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.

            debug: Boolean indicating if debug tracebacks should be returned on errors.
            routes: A list of routes to serve incoming HTTP and WebSocket requests.
            middleware: A list of middleware to run for every request. A starlette application will always
                automatically include two middleware classes. `ServerErrorMiddleware` is added as the very
                outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack.
                `ExceptionMiddleware` is added as the very innermost middleware, to deal with handled
                exception cases occurring in the routing or endpoints.
            exception_handlers: A mapping of either integer status codes, or exception class types onto
                callables which handle the exceptions. Exception handler callables should be of the form
                `handler(request, exc) -> response` and may be either standard functions, or async functions.
            on_startup: A list of callables to run on application startup. Startup handler callables do not
                take any arguments, and may be either standard functions, or async functions.
            on_shutdown: A list of callables to run on application shutdown. Shutdown handler callables do
                not take any arguments, and may be either standard functions, or async functions.
            lifespan: A lifespan context function, which can be used to perform startup and shutdown tasks.
                This is a newer style that replaces the `on_startup` and `on_shutdown` handlers. Use one or
                the other, not both.
        """
        super().__init__(
            debug=debug,
            routes=routes,
            middleware=middleware,
            exception_handlers=exception_handlers,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            lifespan=lifespan,
        )

        async def endpoint(request: Request) -> Response:
            """Endpoint to run the agent with the provided input data."""
            return await handle_ag_ui_request(
                agent,
                request,
                output_type=output_type,
                model=model,
                deps=deps,
                model_settings=model_settings,
                usage_limits=usage_limits,
                usage=usage,
                infer_name=infer_name,
                toolsets=toolsets,
            )

        self.router.add_route('/', endpoint, methods=['POST'], name='run_agent')

init

__init__(
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    *,
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[AgentDepsT]] | None
    ) = None,
    debug: bool = False,
    routes: Sequence[BaseRoute] | None = None,
    middleware: Sequence[Middleware] | None = None,
    exception_handlers: (
        Mapping[Any, ExceptionHandler] | None
    ) = None,
    on_startup: Sequence[Callable[[], Any]] | None = None,
    on_shutdown: Sequence[Callable[[], Any]] | None = None,
    lifespan: (
        Lifespan[AGUIApp[AgentDepsT, OutputDataT]] | None
    ) = None
) -> None

一个 ASGI 应用程序，通过运行代理来处理每个 AG-UI 请求。

请注意，除了注入到实现了 StateHandler 协议的 deps 对象的 state 字段中的 AG-UI 状态外，每个请求的 deps 都是相同的。要为每个请求提供不同的 deps（例如，基于已认证的用户），请改用 pydantic_ai.ag_ui.run_ag_ui 或 pydantic_ai.ag_ui.handle_ag_ui_request。

参数

名称	类型	描述	默认值
`agent`	`AbstractAgent[AgentDepsT, OutputDataT]`	要运行的代理。	必需
`output_type`	`OutputSpec[Any] \| None`	用于本次运行的自定义输出类型，仅当代理没有输出验证器时才可使用 `output_type`，因为输出验证器期望一个与代理输出类型匹配的参数。	`None`
`model`	`Model \| KnownModelName \| str \| None`	用于本次运行的可选模型，如果在创建代理时未设置 `model`，则此项为必需。	`None`
`deps`	`AgentDepsT`	用于本次运行的可选依赖项。	`None`
`model_settings`	`ModelSettings \| None`	用于此模型请求的可选设置。	`None`
`usage_limits`	`UsageLimits \| None`	对模型请求次数或令牌用量的可选限制。	`None`
`usage`	`RunUsage \| None`	可选的初始用量，对于恢复对话或在工具中使用的代理非常有用。	`None`
`infer_name`	`bool`	在未设置代理名称时，是否尝试从调用帧中推断代理名称。	`True`
`toolsets`	`Sequence[AbstractToolset[AgentDepsT]] \| None`	用于本次运行的可选附加工具集。	`None`
`debug`	`bool`	一个布尔值，指示在出错时是否应返回调试追溯信息。	`False`
`routes`	`Sequence[BaseRoute] \| None`	用于处理传入 HTTP 和 WebSocket 请求的路由列表。	`None`
`middleware`	`Sequence[Middleware] \| None`	为每个请求运行的中间件列表。一个 starlette 应用程序总是会自动包含两个中间件类。`ServerErrorMiddleware` 作为最外层的中间件添加，用于处理在整个堆栈中发生的任何未捕获的错误。`ExceptionMiddleware` 作为最内层的中间件添加，用于处理在路由或端点中发生的已处理异常情况。	`None`
`exception_handlers`	`Mapping[Any, ExceptionHandler] \| None`	将整数状态码或异常类类型映射到处理异常的可调用对象的映射。异常处理可调用对象的形式应为 `handler(request, exc) -> response`，可以是标准函数或异步函数。	`None`
`on_startup`	`Sequence[Callable[[], Any]] \| None`	在应用程序启动时运行的可调用对象列表。启动处理可调用对象不接受任何参数，可以是标准函数或异步函数。	`None`
`on_shutdown`	`Sequence[Callable[[], Any]] \| None`	在应用程序关闭时运行的可调用对象列表。关闭处理可调用对象不接受任何参数，可以是标准函数或异步函数。	`None`
`lifespan`	`Lifespan[AGUIApp[AgentDepsT, OutputDataT]] \| None`	一个生命周期上下文函数，可用于执行启动和关闭任务。这是一种较新的风格，取代了 `on_startup` 和 `on_shutdown` 处理程序。请使用其中一种，不要同时使用两者。	`None`

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

def __init__(
    self,
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    *,
    # Agent.iter parameters.
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
    # Starlette parameters.
    debug: bool = False,
    routes: Sequence[BaseRoute] | None = None,
    middleware: Sequence[Middleware] | None = None,
    exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
    on_startup: Sequence[Callable[[], Any]] | None = None,
    on_shutdown: Sequence[Callable[[], Any]] | None = None,
    lifespan: Lifespan[AGUIApp[AgentDepsT, OutputDataT]] | None = None,
) -> None:
    """An ASGI application that handles every AG-UI request by running the agent.

    Note that the `deps` will be the same for each request, with the exception of the AG-UI state that's
    injected into the `state` field of a `deps` object that implements the [`StateHandler`][pydantic_ai.ag_ui.StateHandler] protocol.
    To provide different `deps` for each request (e.g. based on the authenticated user),
    use [`pydantic_ai.ag_ui.run_ag_ui`][pydantic_ai.ag_ui.run_ag_ui] or
    [`pydantic_ai.ag_ui.handle_ag_ui_request`][pydantic_ai.ag_ui.handle_ag_ui_request] instead.

    Args:
        agent: The agent to run.

        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has
            no output validators since output validators would expect an argument that matches the agent's
            output type.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.

        debug: Boolean indicating if debug tracebacks should be returned on errors.
        routes: A list of routes to serve incoming HTTP and WebSocket requests.
        middleware: A list of middleware to run for every request. A starlette application will always
            automatically include two middleware classes. `ServerErrorMiddleware` is added as the very
            outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack.
            `ExceptionMiddleware` is added as the very innermost middleware, to deal with handled
            exception cases occurring in the routing or endpoints.
        exception_handlers: A mapping of either integer status codes, or exception class types onto
            callables which handle the exceptions. Exception handler callables should be of the form
            `handler(request, exc) -> response` and may be either standard functions, or async functions.
        on_startup: A list of callables to run on application startup. Startup handler callables do not
            take any arguments, and may be either standard functions, or async functions.
        on_shutdown: A list of callables to run on application shutdown. Shutdown handler callables do
            not take any arguments, and may be either standard functions, or async functions.
        lifespan: A lifespan context function, which can be used to perform startup and shutdown tasks.
            This is a newer style that replaces the `on_startup` and `on_shutdown` handlers. Use one or
            the other, not both.
    """
    super().__init__(
        debug=debug,
        routes=routes,
        middleware=middleware,
        exception_handlers=exception_handlers,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        lifespan=lifespan,
    )

    async def endpoint(request: Request) -> Response:
        """Endpoint to run the agent with the provided input data."""
        return await handle_ag_ui_request(
            agent,
            request,
            output_type=output_type,
            model=model,
            deps=deps,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            infer_name=infer_name,
            toolsets=toolsets,
        )

    self.router.add_route('/', endpoint, methods=['POST'], name='run_agent')

handle_ag_ui_request `异步`

handle_ag_ui_request(
    agent: AbstractAgent[AgentDepsT, Any],
    request: Request,
    *,
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[AgentDepsT]] | None
    ) = None
) -> Response

通过运行代理并返回流式响应来处理 AG-UI 请求。

参数

名称	类型	描述	默认值
`agent`	`AbstractAgent[AgentDepsT, Any]`	要运行的代理。	必需
`request`	`请求`	包含 AG-UI 运行输入的 Starlette 请求（例如，来自 FastAPI）。	必需
`output_type`	`OutputSpec[Any] \| None`	用于本次运行的自定义输出类型，仅当代理没有输出验证器时才可使用 `output_type`，因为输出验证器期望一个与代理输出类型匹配的参数。	`None`
`model`	`Model \| KnownModelName \| str \| None`	用于本次运行的可选模型，如果在创建代理时未设置 `model`，则此项为必需。	`None`
`deps`	`AgentDepsT`	用于本次运行的可选依赖项。	`None`
`model_settings`	`ModelSettings \| None`	用于此模型请求的可选设置。	`None`
`usage_limits`	`UsageLimits \| None`	对模型请求次数或令牌用量的可选限制。	`None`
`usage`	`RunUsage \| None`	可选的初始用量，对于恢复对话或在工具中使用的代理非常有用。	`None`
`infer_name`	`bool`	在未设置代理名称时，是否尝试从调用帧中推断代理名称。	`True`
`toolsets`	`Sequence[AbstractToolset[AgentDepsT]] \| None`	用于本次运行的可选附加工具集。	`None`

返回

类型	描述
`响应`	一个带有 AG-UI 协议事件的流式 Starlette 响应。

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

async def handle_ag_ui_request(
    agent: AbstractAgent[AgentDepsT, Any],
    request: Request,
    *,
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
) -> Response:
    """Handle an AG-UI request by running the agent and returning a streaming response.

    Args:
        agent: The agent to run.
        request: The Starlette request (e.g. from FastAPI) containing the AG-UI run input.

        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
            output validators since output validators would expect an argument that matches the agent's output type.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.

    Returns:
        A streaming Starlette response with AG-UI protocol events.
    """
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        input_data = RunAgentInput.model_validate(await request.json())
    except ValidationError as e:  # pragma: no cover
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    return StreamingResponse(
        run_ag_ui(
            agent,
            input_data,
            accept,
            output_type=output_type,
            model=model,
            deps=deps,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            infer_name=infer_name,
            toolsets=toolsets,
        ),
        media_type=accept,
    )

run_ag_ui `异步`

run_ag_ui(
    agent: AbstractAgent[AgentDepsT, Any],
    run_input: RunAgentInput,
    accept: str = SSE_CONTENT_TYPE,
    *,
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: (
        Sequence[AbstractToolset[AgentDepsT]] | None
    ) = None
) -> AsyncIterator[str]

使用 AG-UI 运行输入来运行代理，并流式传输 AG-UI 协议事件。

参数

名称	类型	描述	默认值
`agent`	`AbstractAgent[AgentDepsT, Any]`	要运行的代理。	必需
`run_input`	`RunAgentInput`	包含 thread_id、run_id、messages 等的 AG-UI 运行输入。	必需
`accept`	`str`	运行的 accept 标头值。	`SSE_CONTENT_TYPE`
`output_type`	`OutputSpec[Any] \| None`	用于本次运行的自定义输出类型，仅当代理没有输出验证器时才可使用 `output_type`，因为输出验证器期望一个与代理输出类型匹配的参数。	`None`
`model`	`Model \| KnownModelName \| str \| None`	用于本次运行的可选模型，如果在创建代理时未设置 `model`，则此项为必需。	`None`
`deps`	`AgentDepsT`	用于本次运行的可选依赖项。	`None`
`model_settings`	`ModelSettings \| None`	用于此模型请求的可选设置。	`None`
`usage_limits`	`UsageLimits \| None`	对模型请求次数或令牌用量的可选限制。	`None`
`usage`	`RunUsage \| None`	可选的初始用量，对于恢复对话或在工具中使用的代理非常有用。	`None`
`infer_name`	`bool`	在未设置代理名称时，是否尝试从调用帧中推断代理名称。	`True`
`toolsets`	`Sequence[AbstractToolset[AgentDepsT]] \| None`	用于本次运行的可选附加工具集。	`None`

产生

类型	描述
`AsyncIterator[str]`	根据 accept 标头值编码为字符串的流式事件块。

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

async def run_ag_ui(
    agent: AbstractAgent[AgentDepsT, Any],
    run_input: RunAgentInput,
    accept: str = SSE_CONTENT_TYPE,
    *,
    output_type: OutputSpec[Any] | None = None,
    model: Model | KnownModelName | str | None = None,
    deps: AgentDepsT = None,
    model_settings: ModelSettings | None = None,
    usage_limits: UsageLimits | None = None,
    usage: RunUsage | None = None,
    infer_name: bool = True,
    toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
) -> AsyncIterator[str]:
    """Run the agent with the AG-UI run input and stream AG-UI protocol events.

    Args:
        agent: The agent to run.
        run_input: The AG-UI run input containing thread_id, run_id, messages, etc.
        accept: The accept header value for the run.

        output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
            output validators since output validators would expect an argument that matches the agent's output type.
        model: Optional model to use for this run, required if `model` was not set when creating the agent.
        deps: Optional dependencies to use for this run.
        model_settings: Optional settings to use for this model's request.
        usage_limits: Optional limits on model request count or token usage.
        usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
        infer_name: Whether to try to infer the agent name from the call frame if it's not set.
        toolsets: Optional additional toolsets for this run.

    Yields:
        Streaming event chunks encoded as strings according to the accept header value.
    """
    encoder = EventEncoder(accept=accept)
    if run_input.tools:
        # AG-UI tools can't be prefixed as that would result in a mismatch between the tool names in the
        # Pydantic AI events and actual AG-UI tool names, preventing the tool from being called. If any
        # conflicts arise, the AG-UI tool should be renamed or a `PrefixedToolset` used for local toolsets.
        toolset = _AGUIFrontendToolset[AgentDepsT](run_input.tools)
        toolsets = [*toolsets, toolset] if toolsets else [toolset]

    try:
        yield encoder.encode(
            RunStartedEvent(
                thread_id=run_input.thread_id,
                run_id=run_input.run_id,
            ),
        )

        if not run_input.messages:
            raise _NoMessagesError

        raw_state: dict[str, Any] = run_input.state or {}
        if isinstance(deps, StateHandler):
            if isinstance(deps.state, BaseModel):
                try:
                    state = type(deps.state).model_validate(raw_state)
                except ValidationError as e:  # pragma: no cover
                    raise _InvalidStateError from e
            else:
                state = raw_state

            deps = replace(deps, state=state)
        elif raw_state:
            raise UserError(
                f'AG-UI state is provided but `deps` of type `{type(deps).__name__}` does not implement the `StateHandler` protocol: it needs to be a dataclass with a non-optional `state` field.'
            )
        else:
            # `deps` not being a `StateHandler` is OK if there is no state.
            pass

        messages = _messages_from_ag_ui(run_input.messages)

        async with agent.iter(
            user_prompt=None,
            output_type=[output_type or agent.output_type, DeferredToolRequests],
            message_history=messages,
            model=model,
            deps=deps,
            model_settings=model_settings,
            usage_limits=usage_limits,
            usage=usage,
            infer_name=infer_name,
            toolsets=toolsets,
        ) as run:
            async for event in _agent_stream(run):
                yield encoder.encode(event)
    except _RunError as e:
        yield encoder.encode(
            RunErrorEvent(message=e.message, code=e.code),
        )
    except Exception as e:
        yield encoder.encode(
            RunErrorEvent(message=str(e)),
        )
        raise e
    else:
        yield encoder.encode(
            RunFinishedEvent(
                thread_id=run_input.thread_id,
                run_id=run_input.run_id,
            ),
        )

StateHandler

基类：Protocol

代理运行中状态处理器的协议。要求该类是一个数据类，并带有一个 state 字段。

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

@runtime_checkable
class StateHandler(Protocol):
    """Protocol for state handlers in agent runs. Requires the class to be a dataclass with a `state` field."""

    # Has to be a dataclass so we can use `replace` to update the state.
    # From https://github.com/python/typeshed/blob/9ab7fde0a0cd24ed7a72837fcb21093b811b80d8/stdlib/_typeshed/__init__.pyi#L352
    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]

    @property
    def state(self) -> State:
        """Get the current state of the agent run."""
        ...

    @state.setter
    def state(self, state: State) -> None:
        """Set the state of the agent run.

        This method is called to update the state of the agent run with the
        provided state.

        Args:
            state: The run state.

        Raises:
            InvalidStateError: If `state` does not match the expected model.
        """
        ...

state `属性` `可写`

state: State

获取代理运行的当前状态。

StateDeps `数据类`

基类：Generic[StateT]

提供 AG-UI 状态管理。

此类用于管理代理运行的状态。它允许使用特定类型的状态模型来设置代理运行的状态，该模型必须是 BaseModel 的子类。

在运行开始时，状态由 Adapter 通过 state 设置器进行设置。

实现了 StateHandler 协议。

源代码位于 pydantic_ai_slim/pydantic_ai/ag_ui.py

@dataclass
class StateDeps(Generic[StateT]):
    """Provides AG-UI state management.

    This class is used to manage the state of an agent run. It allows setting
    the state of the agent run with a specific type of state model, which must
    be a subclass of `BaseModel`.

    The state is set using the `state` setter by the `Adapter` when the run starts.

    Implements the `StateHandler` protocol.
    """

    state: StateT

pydantic_ai.ag_ui

SSE_CONTENT_TYPE 模块属性

AGUIApp

__init__

handle_ag_ui_request 异步

run_ag_ui 异步

StateHandler

state 属性 可写

StateDeps 数据类

`pydantic_ai.ag_ui`

SSE_CONTENT_TYPE `模块属性`

init

handle_ag_ui_request `异步`

run_ag_ui `异步`

state `属性` `可写`

StateDeps `数据类`