Agent-User Interaction

Agent-User Interaction (AG-UI) Protocol

이 문서에서는 Pydantic AI의 Agent-User Interaction (AG-UI) Protocol 통합을 소개해요. AG-UI는 CopilotKit 팀이 도입한 오픈 표준으로, 프론트엔드 애플리케이션이 AI 에이전트와 통신하는 방식을 표준화하며 스트리밍, 프론트엔드 도구, 공유 상태, 커스텀 이벤트를 지원해요.

출처: 문서

본문

Agent-User Interaction (AG-UI) Protocol은 CopilotKit 팀이 도입한 오픈 표준으로, 프론트엔드 애플리케이션이 AI 에이전트와 통신하는 방식을 표준화하며 스트리밍, 프론트엔드 도구, 공유 상태, 커스텀 이벤트를 지원해요.

Note

AG-UI 통합은 원래 Rocket Science 팀이 구축했으며 Pydantic AI와 CopilotKit 팀과 협력해 기여했어요. Rocket Science 감사합니다!

Installation

유일한 의존성은:

ag-ui-protocol: AG-UI 타입과 인코더 제공.
starlette: FastAPI 같은 프레임워크의 ASGI 요청 처리.

필요한 AG-UI 의존성을 모두 확보하려면 ag-ui extra로 Pydantic AI를 설치할 수 있어요:

Terminal

pip install 'pydantic-ai-slim[ag-ui]'

Terminal

uv add 'pydantic-ai-slim[ag-ui]'

예시를 실행하려면 다음도 필요해요:

uvicorn 또는 다른 ASGI 호환 서버
pip
uv

Terminal

pip install uvicorn

Terminal

uv add uvicorn

Usage

AG-UI 실행 입력 기반의 Pydantic AI 에이전트를 출력으로 스트리밍된 AG-UI 이벤트와 함께 실행하는 방법은 세 가지가 있어요. 가장 유연한 것부터 시작합니다. FastAPI 같은 Starlette 기반 웹 프레임워크를 사용한다면 보통 두 번째 방법을 쓰길 원할 거예요.

AGUIAdapter.run_stream() 메서드는 에이전트와 AG-UI RunAgentInput 객체로 인스턴스화된 AGUIAdapter에서 호출되면 에이전트를 실행하고 AG-UI 이벤트의 스트림을 반환해요. 또한 deps를 포함한 선택적 Agent.iter() 인자를 받아요. Starlette 기반이 아닌 웹 프레임워크(예: Django 또는 Flask)를 사용하거나 입력이나 출력을 어떤 식으로든 수정하려면 이것을 사용하세요.
AGUIAdapter.dispatch_request() 클래스 메서드는 에이전트와 AG-UI 프론트엔드에서 오는 Starlette 요청(예: FastAPI)을 받아, 엔드포인트에서 직접 반환할 수 있는 AG-UI 이벤트의 스트리밍 Starlette 응답을 반환해요. 또한 각 요청마다(예: 인증된 사용자에 기반) 다르게 할 수 있는 deps를 포함한 선택적 Agent.iter() 인자를 받아요. 이것은 AGUIAdapter.from_request(), AGUIAdapter.run_stream(), AGUIAdapter.streaming_response()를 결합한 편의 메서드예요.
AGUIAdapter.dispatch_request()를 호출하는 단일 / 라우트를 가진 독립형 Starlette 앱을 만드세요. 같은 Starlette 앱은 기존 FastAPI 앱의 경로에 마운트할 수 있어요.

실행이 일방 당사자 취소로 끝나면 — ctx.cancel(), AgentRun.cancel(), 또는 서버가 취소 엔드포인트에 연결한 CancellationToken — 어댑터는 열린 텍스트나 도구 이벤트를 닫고 순수한 RUN_FINISHED를 방출해요. AG-UI에는 현재 취소된 결과가 없으므로, 취소가 RUN_ERROR로 보고되지 않아요. RunCancelled.all_messages()에서 재개 가능한 메시지 히스토리를 영속하려면 on_cancel 콜백(아래 run_stream() 예시 참고)을 전달하세요.

클라이언트 연결 끊김은 외부 취소예요

연결을 끊는(또는 요청을 중단하는) 클라이언트는 서버에 일방 당사자 취소가 아닌 외부 asyncio.CancelledError로 보여요(cancelling a run의 두 종류 참고), 그래서 순수한 RUN_FINISHED와 on_cancel은 연결 끊김에서 발화하지 않아요. 이 방식으로 정지 동작을 관찰하려면 스트림을 연결된 상태로 유지하고 별도 취소 엔드포인트에서 트리거된 CancellationToken으로 실행을 일방 당사자 기준으로 취소하세요.

Handle run input and output directly

이 예시는 AGUIAdapter.run_stream()을 사용하고 자체 요청 파싱과 응답 생성을 수행해요. 어떤 웹 프레임워크와도 동작하도록 수정할 수 있어요.

run_ag_ui.py

import json
from http import HTTPStatus

from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import Response, StreamingResponse
from pydantic import ValidationError

from pydantic_ai import Agent, RunCancelled
from pydantic_ai.ui import SSE_CONTENT_TYPE
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-5.2', instructions='Be fun!')

app = FastAPI()


async def on_cancel(cancelled: RunCancelled) -> None:
    messages = cancelled.all_messages()  # the resumable history to persist
    print(f'cancelled after {len(messages)} messages')


@app.post('/')
async def run_agent(request: Request) -> Response:
    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = AGUIAdapter.build_run_input(await request.body())  # (1)
    except ValidationError as e:
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    adapter = AGUIAdapter(agent=agent, run_input=run_input, accept=accept)
    event_stream = adapter.run_stream(on_cancel=on_cancel)  # (2)

    sse_event_stream = adapter.encode_stream(event_stream)
    return StreamingResponse(sse_event_stream, media_type=accept) # (3)

AGUIAdapter.build_run_input()은 요청 본문을 바이트로 받아 AG-UI RunAgentInput 객체를 반환해요. 요청에서 직접 어댑터를 만드는 데 AGUIAdapter.from_request() 클래스 메서드를 사용할 수도 있어요.
AGUIAdapter.run_stream()은 에이전트를 실행하고 AG-UI 이벤트의 스트림을 반환해요. Agent.run_stream_events()와 같은 선택적 인자를 지원하며 deps를 포함해요. 에이전트를 실행하고 Pydantic AI 이벤트의 스트림을 대신 반환하려면 AGUIAdapter.run_stream_native()을 사용할 수도 있으며, AGUIAdapter.transform_stream()으로 AG-UI 이벤트로 변환할 수 있어요.
AGUIAdapter.encode_stream()은 accept 헤더 값에 따라 AG-UI 이벤트의 스트림을 문자열로 인코딩해요. run_stream()이 반환한 AG-UI 이벤트 스트림에서 직접 스트리밍 응답을 생성하려면 AGUIAdapter.streaming_response()를 사용할 수도 있어요.

app은 ASGI 애플리케이션이므로 어떤 ASGI 서버로든 사용할 수 있어요:

Terminal

uvicorn run_ag_ui:app

이렇게 하면 에이전트가 AG-UI 서버로 노출되고, 프론트엔드가 그것에 요청을 보내기 시작할 수 있어요.

Handle a Starlette request

이 예시는 AGUIAdapter.dispatch_request()를 사용해 FastAPI 요청을 직접 처리하고 응답을 반환해요. 이와 유사한 것이 어떤 Starlette 기반 웹 프레임워크에서도 동작해요.

handle_ag_ui_request.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-5.2', instructions='Be fun!')

app = FastAPI()


@app.post('/')
async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent) # (1)

이 메서드는 본질적으로 이전 예시와 같은 일을 하지만, 이미 Starlette/FastAPI 앱을 사용 중일 때 사용하기에 더 편리해요.

app은 ASGI 애플리케이션이므로 어떤 ASGI 서버로든 사용할 수 있어요:

Terminal

uvicorn handle_ag_ui_request:app

이렇게 하면 에이전트가 AG-UI 서버로 노출되고, 프론트엔드가 그것에 요청을 보내기 시작할 수 있어요.

Stand-alone ASGI app

마운트할 Starlette/FastAPI 앱이 이미 없을 때는, AGUIAdapter.dispatch_request()를 호출하는 단일 / 라우트를 가진 최소한의 Starlette 앱을 만드세요:

ag_ui_app.py

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-5.2', instructions='Be fun!')


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

app은 ASGI 애플리케이션이므로 어떤 ASGI 서버로든 사용할 수 있어요:

Terminal

uvicorn ag_ui_app:app

이렇게 하면 에이전트가 AG-UI 서버로 노출되고, 프론트엔드가 그것에 요청을 보내기 시작할 수 있어요.

Design

Pydantic AI AG-UI 통합은 spec의 모든 기능을 지원해요:

통합은 메시지 히스토리, 상태, 사용 가능한 도구를 포함한 요청된 에이전트 실행의 세부사항을 설명하는 RunAgentInput 객체 형태의 메시지를 받아요.

이들은 Pydantic AI 타입으로 변환되어 에이전트의 실행 메서드에 전달돼요. 도구 호출을 포함한 에이전트의 이벤트는 AG-UI 이벤트로 변환되어 Server-Sent Events(SSE)로 호출자에게 스트리밍돼요.

사용자 요청은 필요한 도구와 이벤트에 따라 클라이언트 UI와 Pydantic AI 서버 사이에 여러 왕복이 필요할 수 있어요.

Features

State management

통합은 AG-UI 상태 관리를 완전히 지원해, 에이전트와 프론트엔드 애플리케이션 간의 실시간 동기화를 가능하게 해요.

아래 예시에는 일반 파라미터로 지정된 Pydantic BaseModel을 사용해 RunAgentInput.state에 담긴 상태를 자동으로 검증하는 데 쓸 수 있는 StateDeps 의존성 타입을 사용해 UI와 서버 사이에 공유되는 문서 상태가 있어요.

AG-UI 상태를 가진 커스텀 의존성 타입

AG-UI 상태와 다른 것들을 담기 위해 자체 의존성 타입을 사용하려면, 그것이 StateHandler 프로토콜을 구현해야 해요. 즉, 비선택적 state 필드를 가진 dataclass여야 해요. 이는 Pydantic AI가 매번 새 의존성 객체를 만들어 요청 사이에 상태가 제대로 격리되도록 하기 위함이에요.

state 필드의 타입이 Pydantic BaseModel 하위 클래스라면, 요청의 원시 상태 dict가 자동으로 검증돼요. 그렇지 않으면 의존성 dataclass의 __post_init__ 메서드에서 원시 값을 직접 검증할 수 있어요.

AG-UI 상태가 제공되지만 의존성이 StateHandler를 구현하지 않으면, Pydantic AI는 경고를 발생시키고 상태를 무시해요. 들어오는 상태를 받고 검증하려면 StateDeps 또는 커스텀 StateHandler 구현을 사용하세요.

ag_ui_state.py

from dataclasses import replace

from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-5.2',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
deps = StateDeps(DocumentState())


async def run_agent(request: Request) -> Response:
    # `dispatch_request` mutates `deps.state` from the request, so give each request its own copy.
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

app은 ASGI 애플리케이션이므로 어떤 ASGI 서버로든 사용할 수 있어요:

Terminal

uvicorn ag_ui_state:app --host 0.0.0.0 --port 9000

Tools

AG-UI 프론트엔드 도구는 Pydantic AI 에이전트에 원활하게 제공되어, 프론트엔드 사용자 인터페이스로 풍부한 사용자 경험을 가능하게 해요.

Context

메시지와 함께, AG-UI 클라이언트는 실행과 관련 있다고 간주하는 것들을 설명하는 description/value 쌍의 context 배열을 보낼 수 있어요. 출발 플랫폼, 요청 사용자, 또는 채널의 상설 지침. 모든 항목은 클라이언트가 한 주장이에요 — 원하는 어떤 description/value든 보낼 수 있어요 — 그래서 요청을 설명할 뿐, 누가 요청하는지 확립하지 않아요.

이 항목들은 모델에 자동으로 전달되지 않고, instructions에 들어가지 않아요. Instructions는 운영자 권위를 지녀요 — 그것은 당신의 모델 지침으로 취급되므로 — 클라이언트가 보낸 텍스트로 그것을 만들면 프롬프트 인젝션이 그 권위를 상속하게 해요. 그것을 데이터로 전달하면 그 권위를 부정하지만 안전하게 만들지는 않아요. 그것은 여전히 간접 프롬프트 인젝션 입력이므로, 부작용 도구의 범위를 서버가 확립한 deps로 정하고 재인가하세요. 항목의 description이나 value에서 절대 하지 마세요. Mid-conversation system prompts와 trust model을 참고하세요.

adapter.run_input.context에서 항목을 읽고 모델에 데이터로 전달하세요. 서버가 확립한 사실 — 인증된 사용자, 워크스페이스 — 은 instructions에 들어가야 하는 것들이에요:

ag_ui_context.py

from dataclasses import dataclass

from ag_ui.core import Context
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent, RunContext
from pydantic_ai.ui.ag_ui import AGUIAdapter


@dataclass
class ChannelDeps:
    workspace: str  # (1)
    context: list[Context]  # (2)


agent = Agent('openai:gpt-5.2', deps_type=ChannelDeps)
app = FastAPI()


@agent.instructions
def workspace(ctx: RunContext[ChannelDeps]) -> str:
    return f'You are answering in the {ctx.deps.workspace} workspace.'


@agent.tool
def frontend_context(ctx: RunContext[ChannelDeps]) -> list[str]:
    """Context the frontend says is relevant to this conversation."""
    return [f'{entry.description}: {entry.value}' for entry in ctx.deps.context]


def authenticated_workspace(request: Request) -> str:
    """Whatever your auth layer already established -- a session, a signed token, an API key."""
    ...


@app.post('/')
async def run_agent(request: Request) -> Response:
    adapter = await AGUIAdapter.from_request(request, agent=agent)
    deps = ChannelDeps(workspace=authenticated_workspace(request), context=adapter.run_input.context)
    return adapter.streaming_response(adapter.run_stream(deps=deps))

서버가 확립했으므로 에이전트가 어떻게 행동할지 형성할 수 있어요.

클라이언트가 보냈으므로 모델에 에이전트가 읽을 수 있는 도구 출력으로 도달해요 — 지침으로는 절대 아님.

클라이언트 제공 사실이 에이전트의 행동을 바꾸게 하려면 먼저 그것을 인증하세요. 호출자나 채널을 검증하고, 당신의 서버가 보유한 정책을 조회하고, 그것에서 지침을 쓰세요. 항목 자체는 데이터로 남아요.

모델을 위한 것이 전혀 아닌 것 — Slack 채널 ID, 로케일 — 은 forwardedProps에 담는 것이 낫습니다. 어댑터가 adapter.run_input.forwarded_props로 그대로 통과시켜요. 검증은 모양을 증명할 뿐 정체성을 증명하지 않아요. 사용자가 누구인지, 어떤 테넌트에 있는지, 무엇을 할 수 있는지는 인증된 서버 상태에서 옵니다.

에이전트의 이벤트가 프론트엔드를 서빙하는 요청 밖에서 당신에게 도달할 때는, 그것을 읽을 실행 입력이 전혀 없어요 — AGUIEventStream.thread_id와 run_id가 프로토콜이 요구하는 정체성의 원천으로 인수되는 "Encoding events without a request"를 참고하세요.

context, forwardedProps, parentRunId는 자체 어댑터 프로퍼티가 아니라 run_input에서 직접 읽혀져요. 어댑터의 프로퍼티 — messages, toolset, state, conversation_id, deferred_tool_results — 는 모든 UI 프로토콜이 공유하고 어댑터 자신이 에이전트 실행에 공급하는 개념이에요. 이 세 가지는 AG-UI 특정이며 오직 당신의 코드만 소비하므로, 타입이 프로토콜의 것인 프로토콜 객체에 남아 있어요.

Tool approval (interrupts)

requires_approval=True로 선언된 도구는 AG-UI의 interrupt-aware 실행 수명 주기에 매핑돼요. 모델이 그러한 호출을 제안하면 실행이 일시 중지되고, 어댑터가 outcome.type이 "interrupt"이고 outcome.interrupts[]가 각 보류 승인을 설명하는 RUN_FINISHED 이벤트로 SSE 스트림을 끝내요. 클라이언트는 그 목록에서 승인 UI를 렌더링하고 각 interrupt를 주소화하는 ResumeEntry 항목의 resume[] 배열과 함께 다음 RunAgentInput을 POST해요.

어댑터가 적용하는 매핑(AG-UI Python SDK 필드 이름과 일치):

AG-UI direction

Pydantic AI source / sink

Interrupt.reason

requires_approval=True 도구에 대해 항상 "tool_call"

Interrupt.tool_call_id

제안된 호출의 ToolCallPart.tool_call_id

Interrupt.id

f"int-{tool_call_id}"(재개 시 tool_call_id로 왕복)

Interrupt.metadata

DeferredToolRequests.metadata.get(tool_call_id)

ResumeEntry.payload

{ "approved": bool, "editedArgs"?: object, "reason"?: string }, Interrupt.response_schema에 대해 검증. 검증 실패하는 페이로드는 — 문제의 필드가 approved 자체가 아니더라도 — 거부하고, 잘못 타입된 editedArgs나 reason은 approved=True 옆에도 거부하는 반면, 선택 필드를 생략하거나 null로 보내는 것은 받아들여짐

payload.approved=True

ToolApproved

payload.editedArgs

ToolApproved.override_args(제안된 인자를 완전히 대체)

payload.approved=False

message=payload.reason인 ToolDenied. approved는 필수이므로 생략하면 검증에서 거부되고 그 reason은 사용되지 않아요 — 모델에 reason이 도달하게 하려면 approved: false를 명시적으로 보내세요

status="cancelled"

페이로드와 무관하게 message="Cancelled by user."인 ToolDenied

에이전트가 제안된 호출에서 오류를 내는 대신 깨끗하게 일시 중지할 수 있도록 DeferredToolRequests를 output_type에 포함해야 해요:

ag_ui_tool_approval.py

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent
from pydantic_ai.tools import DeferredToolRequests
from pydantic_ai.ui.ag_ui import AGUIAdapter

agent = Agent('openai:gpt-5.2', output_type=[str, DeferredToolRequests])


@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    """Delete a file. Pauses on a `RUN_FINISHED` interrupt outcome until the user approves."""
    return f'deleted {path}'


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent)


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])

재개된 턴에서 에이전트는 원래 tool_call_id에 대해 도구를 다시 실행하므로, 그 id에 대해 TOOL_CALL_RESULT 이벤트만 방출돼요 — 새 TOOL_CALL_START는 없음. 이는 AG-UI spec이 요구하는 감사 추적을 보존해요.

AG-UI 밖에서도 동작하는 기본 Pydantic AI 원시형은 Deferred tools and human-in-the-loop tool approval를 참고하세요.

버전 요구사항

Interrupts는 ag-ui-protocol >= 0.1.19(PR #1569)이 필요해요. 더 오래된 설치에서는 어댑터가 outcome 없이 순수한 RUN_FINISHED 이벤트를 방출하는 것으로 조용히 폴백하고, 클라이언트가 보내도 resume[]은 무시돼요.

Events

실행이 진행되는 동안 클라이언트로 이벤트를 보내려면 — 예를 들어 장시간 실행 도구의 진행 업데이트 — ctx.emit()로 CustomEvent를 방출하세요:

ag_ui_custom_events.py

from dataclasses import dataclass

from pydantic_ai import Agent, CustomEvent, RunContext

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


@dataclass(kw_only=True)
class SearchIndexProgressEvent(CustomEvent):
    done: int
    total: int


@agent.tool
async def reindex(ctx: RunContext, total: int) -> str:
    for done in range(1, total + 1):
        # Do a unit of work, then tell the frontend how far along we are.
        await ctx.emit(SearchIndexProgressEvent(done=done, total=total))
    return f'Reindexed {total} documents'

각 이벤트는 AG-UI CustomEvent로 그 name과 to_payload()의 결과가 value로 설정되어 클라이언트에 도달해요 — 여기서는 name='search_index_progress'이고 value={'done': 1, 'total': 3}. 이벤트는 도구가 여전히 실행되는 동안 방출될 때 도착해요.

value 모양은 이벤트가 도구 호출 안에서 방출됐는지와 무관하게 같으므로, 한 모양에 맞춰 작성된 프론트엔드는 같은 이벤트 클래스가 나중에 다른 곳에서 방출돼도 깨지지 않아요. 모양을 제어하려면 to_payload()를 재정의하세요 — 프론트엔드가 기대하는 대로 필드 이름을 짓거나, 도구 귀속을 와이어에 올리려면:

ag_ui_custom_event_payload.py

from dataclasses import dataclass
from typing import Any

from pydantic_ai import CustomEvent


@dataclass(kw_only=True)
class SearchIndexPhaseEvent(CustomEvent):
    done: int
    total: int

    def to_payload(self) -> dict[str, Any]:
        return {
            'completed': self.done,
            'total': self.total,
            'toolCallId': self.tool_call_id,
        }

to_payload()에서 AG-UI BaseEvent를 반환하면 그 이벤트가 그대로 전송돼요. 방출된 이벤트가 상태 스냅샷 같은 프로토콜 이벤트를 나를 수도 있어요. ui=False로 선언된 이벤트 클래스는 절대 전달되지 않으므로, 서버 측 소비자 전용 이벤트는 와이어에서 벗어나 있어요. 이 프로세스가 import하지 않은 클래스의 이벤트도 마찬가지인데, 그것의 옵트아웃이 와이어가 아닌 클래스에 있기 때문이에요.

Pydantic AI 도구는 또한 도구 결과에 AG-UI 이벤트를 붙일 수 있어요. BaseEvent(또는 이벤트 목록)를 metadata로 하는 ToolReturn 객체를 반환함으로써요. 방출된 이벤트와 달리, 이것들은 메시지의 일부이며 메시지 히스토리 왕복에서 살아남아요. 프론트엔드가 재구축할 수 있어야 하는 상태 업데이트에 원하는 것이에요. 대가는 도구가 실행되는 동안이 아니라 반환될 때 전송된다는 것이에요.

ag_ui_tool_events.py

from dataclasses import replace

from ag_ui.core import CustomEvent, EventType, StateSnapshotEvent
from pydantic import BaseModel
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route

from pydantic_ai import Agent, RunContext, ToolReturn
from pydantic_ai.ui import StateDeps
from pydantic_ai.ui.ag_ui import AGUIAdapter


class DocumentState(BaseModel):
    """State for the document being written."""

    document: str = ''


agent = Agent(
    'openai:gpt-5.2',
    instructions='Be fun!',
    deps_type=StateDeps[DocumentState],
)
deps = StateDeps(DocumentState())


async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(request, agent=agent, deps=replace(deps))


app = Starlette(routes=[Route('/', run_agent, methods=['POST'])])


@agent.tool
async def update_state(ctx: RunContext[StateDeps[DocumentState]]) -> ToolReturn:
    return ToolReturn(
        return_value='State updated',
        metadata=[
            StateSnapshotEvent(
                type=EventType.STATE_SNAPSHOT,
                snapshot=ctx.deps.state,
            ),
        ],
    )


@agent.tool_plain
async def custom_events() -> ToolReturn:
    return ToolReturn(
        return_value='Count events sent',
        metadata=[
            CustomEvent(
                type=EventType.CUSTOM,
                name='count',
                value=1,
            ),
            CustomEvent(
                type=EventType.CUSTOM,
                name='count',
                value=2,
            ),
        ]
    )

app은 ASGI 애플리케이션이므로 어떤 ASGI 서버로든 사용할 수 있어요:

Terminal

uvicorn ag_ui_tool_events:app --host 0.0.0.0 --port 9000

Protocol version compatibility

Pydantic AI는 0.1.10부터 모든 ag-ui-protocol 릴리스를 지원하며, 그 바닥 이후에 추가된 기능은 업그레이드를 요구하는 대신 설치된 버전에 게이트돼요.

그 게이트는 양방향으로 실행돼요. 나가는 길에 더 오래된 프로토콜 버전이 표현할 수 없는 콘텐츠는 다운그레이드되거나 생략돼요 — 협상된 임계값은 AGUIAdapter.ag_ui_version 참고. 들어오는 길에 설치된 ag-ui-protocol이 클래스가 없는 메시지 role 또는 입력 콘텐츠 type은 태그를 이름짓는 UserWarning과 함께 건너뛰어지고, 요청의 나머지는 실행돼요 — 그래서 서버보다 더 새로운 프로토콜 버전의 프론트엔드가, 설치가 타입이 없는 콘텐츠를 빼고 계속 동작해요. 예를 들어 이미지 첨부를 타입이 지정된 멀티모달 콘텐츠(ag-ui-protocol >= 0.1.15)로 전달하는 게이트웨이는 더 오래된 설치에서 실행되는 에이전트에 동반 텍스트를 여전히 전달해요.

건너뛸 것은 태그만으로 결정돼요. 설치된 모델이 선언하지 않는 어떤 role이나 type 문자열이든 대상이 되므로, "txet"을 잘못 철자하는 클라이언트는 진짜 새 콘텐츠를 보내는 클라이언트와 같은 경고로 건너뛰어져요 — 서버는 그것들을 구분할 방법이 없어요. 스킵은 잘 구성된 항목에 범위가 지정돼요. 메시지는 여전히 모든 AG-UI 메시지 타입이 요구하는 필드인 문자열 id를 지녀야 해요.

그 밖의 모든 것은 여전히 422 Unprocessable Entity로 거부돼요 — 설치가 아는 role 또는 type 아래에서 잘못된 페이로드, 전혀 문자열이 아닌 role이나 type, 유효한 JSON이 아닌 본문. 경고를 보고 콘텐츠가 진짜였다면, ag-ui-protocol을 업그레이드하는 것이 그것이 에이전트에 도달하게 하는 방법이에요.

Trust model

AG-UI의 RunAgentInput.messages는 완전히 클라이언트 제어돼요. AGUIAdapter는 에이전트가 실행되기 전에 신뢰할 수 없는 부분을 제거하는 기본값을 적용해요 — 시스템 프롬프트, 파일 URL 스킴, 업로드된 파일(allow_uploaded_files), 해결되지 않은 도구 호출을 다루는 UI 어댑터 개요의 Trust model for client-submitted messages를 참고하세요. 그 기본값들이 클라이언트 제출 히스토리를 진짜로 만들지는 않아요 — 클라이언트 공급 히스토리의 신뢰 경계를 참고하세요.

Compaction

CompactionPart는 AG-UI 활동 메시지(pydantic_ai_compaction)를 통해 왕복되므로, 프론트엔드가 메시지 히스토리를 쥐고 있을 때 compacted 대화가 계속 동작해요. 프론트엔드가 제출한 compaction 항목은 존중돼요 — 대화가 compacted 상태로 유지 — 두 가지 주의사항과 함께. 첫째, 그것은 시스템 프롬프트 대용으로 절대 신뢰되지 않아요. System prompts and instructions에 따라 적용되는 어떤 프롬프트든 매 요청에 여전히 모델에 도달해요. 둘째, 실행이 서버 측 message_history도 받는다면(서버 측 영속 패턴](https://pydantic.dev/docs/ai/integrations/ui/overview/#trust-model-for-client-submitted-messages)), 프론트엔드 compaction 항목은 무시돼요 — compaction 항목 앞의 모든 것은 모델에서 숨겨지므로, 프론트엔드의 것을 존중하면 서버의 저장된 히스토리를 숨기게 해주겠죠. 트레이드오프와 권장 서버 측 패턴은 Client-held history를 참고하세요.

Assistant message identity

스트리밍된 모든 도구 호출은 그것을 소유한 어시스턴트 메시지를 이름짓는 parentMessageId를 지니고, 그 메시지는 항상 TEXT_MESSAGE_START로 먼저 발표돼요 — 모델의 응답이 도구 호출뿐인 경우에도, 그 경우 메시지는 콘텐츠 없이 즉시 닫혀요. 따라서 이벤트 스트림만으로 대화를 재구축하는 프론트엔드는 메시지가 존재한다는 것을 절대 추론할 필요가 없어요.

도구 호출은 그들이 스트리밍될 때 열려 있는 어시스턴트 메시지에 붙어요. 같은 응답에서 그들 앞에 나타나는 텍스트는 그들의 메시지를 공유하고, 뒤에 나타나는 텍스트는 이후 도구 호출이 대신 붙는 새 메시지를 시작해요. AGUIAdapter.dump_messages는 텍스트와 도구 호출을 같은 방식으로 나눠요. 하지만 그 이상으로 나눠요. compaction 부분은 히스토리가 로드될 때 항상 새 어시스턴트 메시지를 시작하고, reasoning 부분은 ag-ui-protocol 0.1.11부터, 파일 부분은 AGUIAdapter.preserve_file_data 아래에서만. 스트림은 그 어떤 것에 대해서도 나누지 않으므로, 그것 중 하나를 도구 호출과 인터리브하는 응답은 스트리밍된 것보다 더 많은 메시지를 로드해요.

Message ids across round-trips

AGUIAdapter.load_messages는 각 인바운드 메시지의 id를 그것이 만드는 메시지의 예약된 __pydantic_ai__ 키에 유지하고, AGUIAdapter.dump_messages는 그것을 다시 id로 사용하므로, 클라이언트가 보낸 히스토리가 클라이언트가 할당한 id로 돌아와요. 각 ModelRequest 또는 ModelResponse는 id 하나를 지니므로, 연속된 AG-UI 메시지가 하나의 메시지로 병합될 때(병렬 도구 호출의 도구 결과, 또는 시스템 메시지 다음에 사용자 메시지 같은 경우) 마지막 id만 유지되고 그것이 덤프된 마지막 메시지에 붙어요. 나머지는 새 id를 얻어요. 에이전트 실행이 만든 메시지는 유지된 id가 없고 매 덤프마다 새 UUID를 얻어요.

Preserving failed tool outcomes

AG-UI의 ToolCallResultEvent에는 오류나 outcome 필드가 없어요. 암호화된 추론 연속성이 ReasoningEncryptedValueEvent의 의도된 사용이지만, 그것은 메시지나 도구 호출에 encrypted_value를 붙이는 AG-UI의 표준 이벤트이기도 해요. Pydantic AI는 ag-ui-protocol >= 0.1.11을 사용할 때 ToolReturnPart에서 outcome='failed'를 보존하기 위해 네임스페이스된 페이로드로 그 첨부 메커니즘을 사용해요.

클라이언트가 나중의 실행에서 그 메시지들을 다시 보내면, 어댑터가 실패한 outcome을 복원해요. 이것은 히스토리 연속성 메커니즘이에요. ToolMessage.error를 설정하거나 프론트엔드가 결과를 오류로 시각적으로 렌더링한다는 것을 보장하지 않아요. 이전 프로토콜 버전으로 만든 이벤트 스트림은 outcome용 메타데이터 전달자가 없으므로, 그것을 다시 로드하면 도구 결과를 outcome='success'로 재구성해요.

Preserving files across round-trips

AG-UI에는 에이전트 생성 파일(FilePart)이나 UploadedFile 참조를 위한 네이티브 표현이 없으므로, 기본적으로 dump_messages 출력에서 생략돼요. 클라이언트가 다음 요청에 그 활동 메시지를 에코함으로써 완성하는 예약된 pydantic_ai_* 활동 메시지를 통해 왕복하려면 AGUIAdapter.preserve_file_data를 True로 설정하세요. 이것은 표현 옵트인이지 보안 옵트인이 아니에요. 왕복된 활동 메시지에서 재구성된 UploadedFile은 에이전트에 도달하기 전에 여전히 인바운드 allow_uploaded_files 게이트를 받아요.

동작 변경

preserve_file_data는 인바운드 클라이언트 제출 UploadedFile 참조의 존중을 게이트하는 데 사용됐어요. 이제 표현 전용이에요. 앱이 인바운드 업로드 파일을 받아들이기 위해 AGUIAdapter(preserve_file_data=True)를 설정했다면, 두 관심사가 이제 별도 플래그이므로 allow_uploaded_files=True도 설정해야 해요.

System prompts and instructions

Pydantic AI는 모델에 안내를 제공하는 두 가지 방법을 지원해요. system_prompt(메시지 히스토리에 SystemPromptPart로 저장)와 instructions(매 요청에 새로 주입, 절대 영속되지 않음). 서버 측을 제어한다면 instructions가 권장 기본값이에요.

이 섹션의 나머지는 system_prompt를 사용할 때만 관련돼요. instructions만 사용한다면 구성할 것이 없어요 — AG-UI 메시지 히스토리와 무관하게 항상 적용되니까요.

system_prompt의 경우 AGUIAdapter의 manage_system_prompt 파라미터로 누가 그것을 소유하는지 선택하세요:

'server'(기본값): 에이전트의 구성된 system_prompt가 권위가 있어요. 프론트엔드가 보낸 어떤 SystemMessage든 경고와 함께 제거되고(악의적인 클라이언트가 공들인 API 요청으로 임의 지침을 주입할 수 있으므로), 에이전트의 자체 시스템 프롬프트는 ReinjectSystemPrompt capability를 통해 첫 요청의 앞에 재주입돼요.
'client': 프론트엔드가 시스템 프롬프트를 소유해요. 프론트엔드 SystemMessage는 그대로 보존되고, 에이전트의 구성된 system_prompt는 주입되지 않아요 — 원하면 매 턴마다 보내는 것은 전적으로 호출자에게 달려 있어요. 구성된 것으로 폴백하는 동작에 옵트인하려면 ReinjectSystemPrompt capability를 에이전트에 추가하세요.

ag_ui_client_managed_system_prompt.py

from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import Response

from pydantic_ai import Agent
from pydantic_ai.ui.ag_ui import AGUIAdapter

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

app = FastAPI()


@app.post('/')
async def run_agent(request: Request) -> Response:
    return await AGUIAdapter.dispatch_request(
        request, agent=agent, manage_system_prompt='client'
    )

Channels

AG-UI 위에 노출한 에이전트는 Slack이나 다른 메시징 플랫폼의 봇도 구동할 수 있어요. CopilotKit Channels SDK가 플랫폼 이벤트를 받고, AG-UI 위에서 에이전트를 실행하며, 그 응답을 네이티브 플랫폼 콘텐츠로 렌더링해요.

Note

CopilotKit이 플랫폼 설정과 배포 지침을 유지해요. 이 섹션은 Pydantic AI 통합을 보여줘요. 전체 워크스루는 CopilotKit Slack for Pydantic AI 가이드를 사용하세요.

How it fits together

관리되는 Slack의 경우 CopilotKit Intelligence가 Slack 자격 증명을 쥐고 각 턴을 @copilotkit/channels로 빌드된 장시간 실행 Node 프로세스에 전달해요. 그 프로세스는 AG-UI 위에서 Pydantic AI 서버에 대화를 보내고 스트리밍된 응답을 Slack에 반환해요.

Slack  ──►  CopilotKit Intelligence  ──►  channel process (Node)  ──►  Pydantic AI server (AG-UI)

채널 프로세스를 만들고 그 AG-UI 클라이언트를 Pydantic AI 서버로 가리키려면 CopilotKit 가이드를 따르세요. 프로세스가 플랫폼 이벤트를 처리할 때, 그것은 트리거 메시지를 에이전트에 전달하고 플랫폼과 사용자 세부사항을 AG-UI context 항목으로 붙일 수 있어요.

AG-UI context는 클라이언트 제공 데이터이므로, Pydantic AI는 그것을 모델 프롬프트에 자동으로 넣지 않아요. 채널 프로세스를 위의 ag_ui_context.py 서버와 나란히 실행하세요. 그것이 adapter.run_input.context를 읽고, 인증된 워크스페이스 데이터를 분리하며, 채널 항목을 지침으로 취급하는 대신 frontend_context 도구로 노출해요.

Terminal

uvicorn ag_ui_context:app

CopilotKit 가이드에 설명된 대로 채널 프로세스를 시작하세요. 서버리스 요청 핸들러가 자체 영속 게이트웨이 연결을 소유할 수 없으므로 장시간 실행 호스트가 필요해요.

Slack

관리되는 Slack 연결은 CopilotKit Intelligence에서 구성되며, Slack 앱을 만들고 그 자격 증명을 쥐도록 안내해요. 실제 워크스페이스에서 봇을 언급하고 직접 메시지를 테스트해 플랫폼 연결, 게이트웨이 리스너, AG-UI 서버, 응답 경로가 함께 동작하는지 검증하세요.

Other platforms

관리 및 개발자 운영 연결은 설정과 지원이 다릅니다. 현재 관리 플랫폼, 직접 어댑터, 프로바이더별 가이드는 Channels SDK 참조를 참고하세요.

Note

CopilotKit Intelligence는 관리되는 대화 히스토리를 재구성하지만, SDK 워크플로우 상태와 대화형 콜백 스냅샷은 기본적으로 인메모리 저장소를 사용해요. 재시작 안전 상태나 상호작용을 약속하기 전에 영속 저장소를 구성하세요. Persistence and scaling 참고.

Examples

더 많은 예시는 pydantic_ai_examples.ag_ui를 참고하세요. AG-UI Dojo와 함께 사용할 서버를 포함해요.

위 예시들은 SSE 위에 AG-UI를 서빙하며, 여기서 각 실행은 자체 요청이에요. pydantic-ai-ws-agent는 chanx를 사용해 같은 프로토콜을 WebSocket 위에 나르는 커뮤니티 예시예요. 연결이 양방향이고 장수하므로, 한 실행이 여러 브라우저 탭에 한 번에 스트리밍되고 그 중 아무 탭이나 도구 호출을 승인할 수 있어요. 메시지 히스토리는 서버에 유지돼요. 그것이 구축한 어댑터는 chanx-kit의 복사본 컴포넌트예요.

Agent-User Interaction