UI Event Streams

UI Event Streams

이 문서에서는 채팅 앱이나 AI 에이전트용 대화형 프론트엔드를 만들 때 사용하는 UI 이벤트 스트림 프로토콜을 소개해요. 백엔드는 프론트엔드에서 에이전트 실행 입력(채팅 메시지 또는 전체 메시지 히스토리)을 받고, 사용자가 실시간으로 무슨 일이 일어나는지 알도록 에이전트의 이벤트(텍스트, thinking, 도구 호출 등)를 프론트엔드로 스트리밍해야 해요.

출처: 문서

본문

채팅 앱이나 AI 에이전트용 대화형 프론트엔드를 만들고 있다면, 백엔드는 프론트엔드에서 에이전트 실행 입력(채팅 메시지 또는 완전한 메시지 히스토리)을 받고, 사용자가 실시간으로 무슨 일이 일어나는지 알도록 에이전트의 이벤트(텍스트, thinking, 도구 호출)를 프론트엔드로 스트리밍해야 해요.

프론트엔드가 Pydantic AI의 ModelRequest와 AgentStreamEvent를 직접 사용할 수 있지만, 보통은 프론트엔드 프레임워크가 네이티브 지원하는 UI 이벤트 스트림 프로토콜을 사용하고 싶을 거예요.

실행 중간에 자체 데이터를 프론트엔드로 푸시하려면(장시간 실행 도구의 진행 업데이트처럼) 커스텀 이벤트를 방출하세요. 아래 두 프로토콜은 서버 측에서만 원하는 이벤트에 대해 그 클래스가 ui=False로 선언되지 않는 한, 자체 커스텀 데이터 이벤트로 전달해요.

Pydantic AI는 두 UI 이벤트 스트림 프로토콜을 네이티브 지원해요:

• Agent-User Interaction (AG-UI) Protocol
• Vercel AI Data Stream Protocol

이 통합들은 추상 UIAdapter 클래스의 하위 클래스로 구현되므로, 다른 UI 이벤트 스트림 프로토콜과 통합하는 데 참고 자료 역할도 해요.

Usage

프로토콜별 UIAdapter 하위 클래스(즉 AGUIAdapter 또는 VercelAIAdapter)는 프론트엔드에서 받은 에이전트 실행 입력을 Agent.run_stream_events()의 인자로 변환하고, 에이전트를 실행한 다음, Pydantic AI 이벤트를 프로토콜별 이벤트로 변환하는 일을 담당해요. 이벤트 스트림 변환은 프로토콜별 UIEventStream 하위 클래스가 처리하며, 에이전트의 이벤트가 프론트엔드를 서빙하는 요청 밖에서 당신에게 도달하는 경우(아래 "Encoding events without a request"에서 다룸)가 아니라면 보통 직접 사용하지 않아요.

FastAPI 같은 Starlette 기반 웹 프레임워크를 사용한다면 엔드포인트 함수에서 UIAdapter.dispatch_request() 클래스 메서드를 사용해 요청을 직접 처리하고 프로토콜별 이벤트의 스트리밍 응답을 반환할 수 있어요. 이는 다음 섹션에서 보여드려요.

Starlette 기반이 아닌 웹 프레임워크(예: Django 또는 Flask)를 사용하거나 입력·출력에 세밀한 제어가 필요하면 UIAdapter 인스턴스를 만들고 그 메서드를 직접 사용할 수 있어요. 이는 아래 "Advanced Usage" 섹션에서 보여드려요.

Usage with Starlette/FastAPI

UIAdapter.dispatch_request()는 요청 외에 에이전트, Agent.run_stream_events()와 같은 선택적 인자, 성공 실행용 선택적 on_complete 콜백, 그리고 일방 당사자 취소(클라이언트 연결 끊김은 외부 취소이며 트리거하지 않음)된 실행에 대해 RunCancelled를 받는 선택적 on_cancel 콜백을 받아요. 두 콜백 모두 선택적으로 추가 프로토콜별 이벤트를 생성할 수 있어요.

Note

이 예시들은 VercelAIAdapter를 사용하지만, 같은 패턴이 모든 UIAdapter 하위 클래스에 적용돼요.

dispatch_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.vercel_ai import VercelAIAdapter

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

app = FastAPI()

@app.post('/chat')
async def chat(request: Request) -> Response:
    return await VercelAIAdapter.dispatch_request(request, agent=agent)

Advanced Usage

Starlette 기반이 아닌 웹 프레임워크(예: Django 또는 Flask)를 사용하거나 입력·출력에 세밀한 제어가 필요하면, UIAdapter 인스턴스를 만들고 그 메서드를 직접 사용할 수 있어요. 위에서 본 UIAdapter.dispatch_request() 클래스 메서드와 같은 일을 달성하도록 체인할 수 있어요:

1. UIAdapter.build_run_input() 클래스 메서드는 요청 본문을 바이트로 받아 프로토콜별 실행 입력 객체를 반환하며, 그것을 agent와 함께 UIAdapter() 생성자에 전달할 수 있어요.
   • 또한 UIAdapter.from_request() 클래스 메서드를 사용해 Starlette/FastAPI 요청에서 직접 어댑터를 구축할 수 있어요.
2. UIAdapter.run_stream() 메서드는 에이전트를 실행하고 프로토콜별 이벤트의 스트림을 반환해요. Agent.run_stream_events()와 같은 선택적 인자를 지원하며, on_complete와 on_cancel 콜백을 포함해요.
   • 또한 UIAdapter.run_stream_native()을 사용해 에이전트를 실행하고 Pydantic AI 이벤트의 스트림을 대신 반환할 수 있으며, 이후 UIAdapter.transform_stream()을 사용해 프로토콜별 이벤트로 변환할 수 있어요.
3. UIAdapter.encode_stream() 메서드는 프로토콜별 이벤트의 스트림을 SSE(HTTP Server-Sent Events) 문자열로 인코딩하며, 그런 다음 스트리밍 응답으로 반환할 수 있어요.
   • 또한 UIAdapter.streaming_response()을 사용해 run_stream()이 반환한 프로토콜별 이벤트 스트림에서 직접 Starlette/FastAPI 스트리밍 응답을 생성할 수 있어요.

Note

이 예시는 FastAPI를 사용하지만, 어떤 웹 프레임워크와도 동작하도록 수정할 수 있어요.

run_stream.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
from pydantic_ai.ui import SSE_CONTENT_TYPE
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

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

app = FastAPI()


@app.post('/chat')
async def chat(request: Request) -> Response:
    media_type = request.headers.get('content-type', '').split(';')[0].strip().lower()
    if media_type != 'application/json':  # (1)
        return Response(status_code=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)

    accept = request.headers.get('accept', SSE_CONTENT_TYPE)
    try:
        run_input = VercelAIAdapter.build_run_input(await request.body())
    except ValidationError as e:
        return Response(
            content=json.dumps(e.json()),
            media_type='application/json',
            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
        )

    adapter = VercelAIAdapter(agent=agent, run_input=run_input, accept=accept)
    event_stream = adapter.run_stream()

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

build_run_input()은 바이트를 받으므로 from_request()가 하는 media-type 검사를 적용할 수 없어요 — 그것이 무엇을 위한 것인지는 the trust model를 참고하세요. 여기처럼 본문을 직접 읽을 때마다 자체 프레임워크의 관용구로 하세요.

Running the agent elsewhere

어댑터는 에이전트가 프론트엔드를 서빙하는 요청 안에서 실행된다고 가정해요. 꼭 그래야 하는 것은 아니에요. 에이전트가 durable execution 워커, 백그라운드 작업, 또는 다른 서비스에서 실행된다면, 어댑터의 두 작업은 단순히 그 경계를 가로질러 분리돼요. 요청 측은 실행 입력을 만들고 자신이 돌려받는 이벤트를 프로토콜 이벤트로 변환하며, 에이전트를 실행하는 쪽은 같은 요청 본문에서 실행 인자를 만듭니다.

요청 측은 절대 run_stream()을 호출하지 않아요. 에이전트의 이벤트가 전송을 통해 도착하는 대로 UIAdapter.transform_stream()에 넘겨줘요(위 예시의 media-type 및 검증 처리는 여전히 적용되며 여기서 생략):

remote_run_handler.py

async def chat(request: Request) -> Response:
    body = await request.body()
    events = start_the_run_somewhere_else(body)  # an async iterator of Pydantic AI events

    adapter = VercelAIAdapter(agent=agent, run_input=VercelAIAdapter.build_run_input(body))
    return adapter.streaming_response(adapter.transform_stream(events))

전송은 실행의 AgentRunResultEvent와 함께 AgentStreamEvent들도 전달해야 해요 — 그것이 프로토콜을 닫고 on_complete가 받는 것이므로 — 그러니 모델의 이벤트만이 아니라 Agent.run_stream_events()가 생성하는 것을 나르는 전송이 필요해요.

Temporal의 Workflow Streams는 그러한 전송 중 하나이며, 작업된 예시로 읽을 가치가 있어요. 워크플로우가 큐이고, 이벤트가 영속적이고 오프셋 주소 지정되며, 재연결하는 프론트엔드가 시작하지 않은 실행에 다시 붙을 수 있어요.

Encoding events without a request

에이전트가 프론트엔드를 서빙하는 요청 안에서 실행되지 않는다면 — 그 이벤트가 durable execution 워크플로우, 큐, 또는 websocket 팬아웃 같은 자체 전송을 통해 API 엣지에 도달 — 실행 입력을 만들 요청 본문도 없고, 에이전트를 실행할 UIAdapter도 없어요. 대신 프로토콜별 UIEventStream 하위 클래스를 단독으로 사용하세요. 그것은 에이전트의 이벤트를 변환·인코딩하며 실행 입력을 받지 않아요.

encode_events.py

from collections.abc import AsyncIterator

from pydantic_ai.ui import NativeEvent
from pydantic_ai.ui.vercel_ai import VercelAIEventStream


async def encode_events(events: AsyncIterator[NativeEvent]) -> AsyncIterator[str]:
    event_stream = VercelAIEventStream()
    async for sse_event in event_stream.encode_stream(event_stream.transform_stream(events)):
        yield sse_event

이벤트 스트림 인스턴스는 한 실행의 상태를 진행하면서 지녀요(현재 메시지 ID, 스트리밍 중인 부분, 기다리는 도구 호출). 실행별로 새 것을 만드세요.

AG-UI 프로토콜은 모든 실행을 프론트엔드에 식별해요. RUN_STARTED와 RUN_FINISHED 이벤트는 thread ID와 run ID를 지니며, AGUIEventStream은 run input이 있을 때 그것을 읽고, 이후 재정의할 ID를 전달하면 경고해요. run input이 없으면 자체 전송이 이미 대화와 실행에 할당한 ID를 전달하세요:

encode_ag_ui_events.py

from pydantic_ai.ui.ag_ui import AGUIEventStream

event_stream = AGUIEventStream(thread_id='conversation-123', run_id='workflow-456')

각각 기본값은 새 UUID이며, 스트림이 구성될 때마다 새로 만들어져요. 하나 이상의 실행에 걸친 대화 — 무엇보다 도구 승인 후 재개된 실행 — 은 프론트엔드가 실행을 상관시키도록 자체 thread_id를 전달해야 해요. 재생 가능한 durable execution 워크플로우 코드 안에서 스트림을 구성하면 기본값이 매 재생마다 다시 만들어지므로 결정성 위험이 되기도 해요. 그러니 거기서 명시적 thread_id와 run_id를 전달하세요. 에이전트 실행 자체의 conversation_id와 run_id를 전달하면 프로토콜의 정체성이 실행의 트레이스와 정렬돼요. 요청 경로에서 AGUIAdapter는 thread ID만 정렬해요. run input의 thread ID를 에이전트의 conversation_id에 매핑하는 반면, 프로토콜의 run ID는 클라이언트가 보낸 것으로 유지되고 에이전트 실행의 run_id에 절대 연결되지 않으므로, 둘은 다르죠.

Trust model for client-submitted messages

UI 어댑터 엔드포인트는 인증 경계가 아니에요. AG-UI와 Vercel AI 프로토콜 둘 다 클라이언트가 각 요청에 전체 대화 히스토리를 전송하도록 설계되었으므로, 프로토콜의 message_history에 있는 모든 것 — 어시스턴트 메시지, 도구 호출, 파일 URL, 도구 결과 — 은 호출자의 통제 아래 있어요. 어댑터 엔드포인트를 내부 백엔드 서비스로 취급하고, 자체 인증된 라우트 핸들러 안에서 실행하세요. 두 프로토콜이 가정하는 배포 모델에 대해서는 AG-UI 보안 고려 사항 페이지를 참고하세요.

엔드포인트 인증은 누가 호출할 수 있는지 결정하지, 무엇이 호출을 일으켰는지 결정하지 않아요. 그 인증이 브라우저가 스스로 붙이는 쿠키나 다른 자격 증명이라면, 사용자가 다른 곳에서 열어 둔 페이지가 그들을 대신해 라우트에 post할 수 있어요. 실행이 시작되고 도구가 실행되며, 공격자는 응답을 읽을 필요조차 없어요. 브라우저가 preflight 없이 위조할 수 있는 요청이 에이전트에 도달하지 않도록, 어댑터는 application/json 요청 본문만 받아들이고 본문을 읽기 전에 그 외의 것은 415로 답해요. 두 프로토콜의 SDK 전송은 그 콘텐츠 유형을 보내므로, 다른 오리진의 프론트엔드는 전과 똑같이 preflight되고 자체 CORS 정책에 의해 허용돼요. 자체 CSRF 보호를 지닌 라우트에서 검사를 건너뛰거나 집합을 넓히려면, UIAdapter.from_request()와 UIAdapter.dispatch_request()의 allowed_content_types 인자를 사용하세요. 그것은 CSRF 전략이 아니라 하나의 제어예요. 주변 자격 증명을 지닌 라우트에서는 프레임워크가 제공하는 무엇이든 그것과 짝지으세요.

어댑터는 권위 있는 상태가 당신 쪽에 유지되도록 몇 가지 기본값을 적용해요:

• 시스템 프롬프트 -- 클라이언트가 제출한 SystemPromptPart는 기본적으로 제거되고 에이전트의 구성된 프롬프트로 대체돼요. UIAdapter.manage_system_prompt로 제어하세요. 각 어댑터의 문서를 참고하세요.
• 매달린 도구 호출 -- 클라이언트 제출 히스토리가 해결되지 않은 ToolCallPart가 있고 일치하는 deferred_tool_results가 없는 ModelResponse로 끝나면, 모델이 절대 방출하지 않은 해결되지 않은 도구 호출을 에이전트가 실행하지 않도록 best-effort 기본값으로 경고와 함께 도구 호출이 버려져요. human-in-the-loop 재개를 위해 실행 메서드에 명시적 deferred_tool_results를 전달하세요 — 그 결과로 해결된 도구 호출은 유지돼요(아래 경고 참고).
• 파일 URL 스킴 -- 클라이언트 제출 메시지의 FileUrl 부분에는 기본적으로 http와 https만 허용돼요. s3://나 gs:// 같은 비 HTTP 스킴은 프로바이더가 서버의 IAM 역할이나 서비스 계정으로 객체를 가져오게 하므로 버려져요. UIAdapter.allowed_file_url_schemes 참고.
• 파일 URL 다운로드 모드 -- FileUrl.force_download 값 중 False가 아닌 것은 클라이언트 제출 메시지에서 기본적으로 False로 리셋돼요. 이는 클라이언트가 서버에 URL을 강제로 가져오게 하거나, 'allow-local'로 SSRF 사설 IP 차단을 옵트아웃하게 하는 것을 방지해요. 프론트엔드를 감사한 후 UIAdapter.allowed_file_url_force_download로 추가 값을 옵트인하세요.
• 업로드된 파일 -- 클라이언트가 제출한 UploadedFile 부분은 비 HTTP FileUrl처럼 기본적으로 버려져요. 서버가 자체 자격 증명으로 프로바이더의 파일 저장소 API에 대해 해석하니까요. 프론트엔드를 감사한 후 UIAdapter.allow_uploaded_files를 True로 설정해 존중하세요. 이것은 순수 인바운드 보안 설정이에요. 에이전트가 만드는 파일 콘텐츠는 항상 클라이언트로 나가는 길에 직렬화돼요.

도구 승인과 결과는 클라이언트가 제출해요

human-in-the-loop 재개 경로에서 실행 메서드에 전달된 DeferredToolResults — 승인, 거부, 외부 실행 도구 결과 — 는 히스토리와 함께 클라이언트가 제출하며, 어댑터는 승인된 도구 호출이 서버가 실제로 발행한 것인지 검증하지 않아요. 엔드포인트에 도달할 수 있는 클라이언트는 따라서 자기 자신이 만든 도구 호출을 승인할 수 있으며, 승인 게이트된 도구를 위한 것도 포함해요. 승인은 인간 서명 없이 모델 이 행동하는 것을 방지해요. 이것은 클라이언트에 대한 인가 경계가 아니에요.

도구 함수에 도달하는 모든 호출에 대해 엔드포인트를 인증하고(위처럼), 의존성에 담긴 인증된 사용자에 대해 — 클라이언트가 공급한 승인이 아니라 — 도구 함수 자체 안에서 민감한 작업의 인가를 강제하세요. 또는 일시 중단된 실행을 서버 측에 영속하고 클라이언트의 것을 존중하는 대신 자체 deferred_tool_results를 전달하세요.

더 엄격한 대화 무결성(예: 이전 어시스턴트 턴과 도구 반환이 서버가 실제로 만든 것과 일치하는지 보장)을 위해, 히스토리를 thread/session ID로 키하여 서버 측에 영속하고 message_history를 통해 어댑터에 전달하세요 — 호출자가 공급한 히스토리는 서버 측 영속에서 온 것으로 신뢰되며 이 정화의 대상이 아니에요.

이 기본값들은 위조된 히스토리가 도달할 수 있는 것을 좁히지만, 히스토리를 제출할 수 있는 클라이언트는 항상 위조할 수 있어요. 이것이 애플리케이션의 인가 모델에 무엇을 의미하는지는 클라이언트 공급 히스토리의 신뢰 경계를 참고하세요.

더 알아보기 (Learn more)

• AG-UI 프로토콜
• Vercel AI Data Stream 프로토콜