Vercel AI Data Stream Protocol

Vercel AI Data Stream Protocol

이 문서에서는 Pydantic AI가 Vercel AI Data Stream Protocol을 네이티브 지원해 useChat 같은 AI SDK UI 훅을 사용하는 프론트엔드로 에이전트 실행 입력을 받고 이벤트를 스트리밍하는 방법을 알려드려요. 사전 빌드된 UI 컴포넌트용으로 AI Elements를 선택적으로 사용할 수 있어요.

출처: 문서

본문

Pydantic AI는 Vercel AI Data Stream Protocol을 네이티브 지원해 useChat 같은 AI SDK UI 훅을 사용하는 프론트엔드로부터 에이전트 실행 입력을 받고 이벤트를 스트리밍해요. AI Elements를 사전 빌드된 UI 컴포넌트용으로 선택적으로 사용할 수 있어요.

Note

기본적으로 어댑터는 역호환성을 위해 AI SDK v5를 대상으로 해요. AI SDK v6에서 도입된 기능을 사용하려면 어댑터에 sdk_version=6을 설정하세요.

Usage

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

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

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

Usage with Starlette/FastAPI

VercelAIAdapter.dispatch_request()는 요청 외에 에이전트, Agent.run_stream_events()와 같은 선택적 인자, 성공 실행용 선택적 on_complete 콜백, 취소된 실행용 선택적 on_cancel 콜백을 받아요. 두 콜백 모두 선택적으로 추가 Vercel AI 이벤트를 생성할 수 있어요.

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)를 사용하거나 입력·출력에 세밀한 제어가 필요하면, VercelAIAdapter 인스턴스를 만들고 그 메서드를 직접 사용할 수 있어요. 위에서 본 VercelAIAdapter.dispatch_request() 클래스 메서드와 같은 일을 달성하도록 체인할 수 있어요:

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

Note

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

Cancellation

실행이 일방 당사자 취소로 끝나면 — 도구의 ctx.cancel(), AgentRun.cancel(), 또는 서버가 취소 엔드포인트에 연결한 CancellationToken — 어댑터가 Vercel abort 청크를 방출해요. useChat은 부분 메시지를 유지하고 오류 상태에 들어가는 대신 isAbort를 onFinish에 보고해요. 아래 예시처럼 on_cancel 콜백을 전달해 재개 가능한 메시지 히스토리를 영속하세요.

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

클라이언트에서 stop()을 호출하면 브라우저의 요청이 중단되며, 서버는 그것을 일방 당사자 취소가 아닌 _연결 끊김_으로 봐요. 그것은 외부 asyncio.CancelledError로 실행을 무너뜨리므로(cancelling a run의 두 종류 참고), abort 청크가 방출되지 않고 on_cancel도 발화하지 않아요 — 그리고 클라이언트는 어쨌든 끊겼어요. 정지 동작에서 abort 청크를 얻고 on_cancel을 실행하려면 스트림을 연결된 상태로 유지하고 실행을 일방 당사자 기준으로 취소하세요. 실행에 CancellationToken을 주고 token.cancel()을 호출하는 별도 엔드포인트(예: POST /chat/{id}/cancel)를 노출하세요.

Note

아래의 인메모리 토큰 레지스트리는 단일 서버 프로세스 또는 끈적한(sticky) 라우팅이 필요해요. 다중 워커 배포에서는 메시지 브로커 같은 공유 조정을 사용해 취소 요청을 실행을 소유한 워커로 라우팅하세요.

run_stream.py

import json
from collections.abc import AsyncIterator
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, CancellationToken, RunCancelled
from pydantic_ai.ui import SSE_CONTENT_TYPE
from pydantic_ai.ui.vercel_ai import VercelAIAdapter

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

app = FastAPI()

cancellation_tokens: dict[str, CancellationToken] = {}


async def on_cancel(cancelled: RunCancelled) -> None:
    messages = cancelled.all_messages()  # (1)
    print(f'cancelled after {len(messages)} messages')


@app.post('/chat/{chat_id}')
async def chat(chat_id: str, request: Request) -> Response:
    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)
    cancellation_token = CancellationToken()
    cancellation_tokens[chat_id] = cancellation_token
    event_stream = adapter.run_stream(
        cancellation_token=cancellation_token, on_cancel=on_cancel
    )

    async def encode_stream() -> AsyncIterator[str]:
        try:
            async for event in adapter.encode_stream(event_stream):
                yield event
        finally:
            if cancellation_tokens.get(chat_id) is cancellation_token:
                cancellation_tokens.pop(chat_id, None)

    return StreamingResponse(encode_stream(), media_type=accept)


@app.post('/chat/{chat_id}/cancel', status_code=HTTPStatus.NO_CONTENT)
async def cancel_chat(chat_id: str) -> None:
    if token := cancellation_tokens.get(chat_id):
        token.cancel()

재개 가능한 히스토리 — 이후 실행에 message_history로 전달해 대화를 재개하세요.

Data Chunks

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

vercel_ai_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 FileUploadProgressEvent(CustomEvent):
    done: int
    total: int


@agent.tool
async def upload_files(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(FileUploadProgressEvent(done=done, total=total))
    return f'Uploaded {total} files'

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

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

vercel_ai_custom_event_payload.py

from dataclasses import dataclass
from typing import Any

from pydantic_ai import CustomEvent


@dataclass(kw_only=True)
class FileUploadPhaseEvent(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()에서 데이터 운반 청크(아래 참고)를 반환하면 그 청크가 그대로 전송돼요. ui=False로 선언된 이벤트 클래스는 절대 전달되지 않으므로, 서버 측 소비자 전용 이벤트는 와이어에서 벗어나 있어요. 이 프로세스가 import하지 않은 클래스의 이벤트도 마찬가지인데, 그것의 옵트아웃이 와이어가 아닌 클래스에 있기 때문이에요.

Pydantic AI 도구는 또한 도구 결과에 Vercel AI data stream chunks를 붙일 수 있어요. 데이터 운반 청크(또는 청크 목록)를 metadata로 하는 ToolReturn 객체를 반환함으로써요. 지원되는 청크 타입은 DataChunk, SourceUrlChunk, SourceDocumentChunk, FileChunk이에요. 방출된 이벤트와 달리, 이것들은 메시지의 일부이며 메시지 히스토리 왕복에서 살아남아요. 이것은 프론트엔드가 재구축할 수 있어야 하는 데이터(답변 뒤의 소스 URL 등)에 원하는 것이에요. 대가는 도구가 실행되는 동안이 아니라 반환될 때 전송된다는 것이에요.

vercel_ai_tool_chunks.py

from pydantic_ai import Agent, ToolReturn
from pydantic_ai.ui.vercel_ai.response_types import DataChunk, SourceUrlChunk

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


@agent.tool_plain
async def search_docs(query: str) -> ToolReturn:
    return ToolReturn(
        return_value=f'Found 2 results for "{query}"',
        metadata=[
            SourceUrlChunk(
                source_id='doc-1',
                url='https://example.com/docs/intro',
                title='Introduction',
            ),
            DataChunk(
                type='data-search-results',
                data={'query': query, 'count': 2},
            ),
        ],
    )

Note

StartChunk, FinishChunk, StartStepChunk, FinishStepChunk 같은 프로토콜 제어 청크는 자동으로 걸러져요 — 위 나열된 네 가지 데이터 운반 청크 타입만 스트림으로 전달되고 dump_messages에서 보존돼요.

Files from client-side tools

Vercel AI SDK 클라이언트 측 도구는 브라우저에서 실행되고 결과를 서버로 다시 제출하며, Pydantic AI는 그것을 외부 도구 호출로 해석해요. 그러한 도구는 Pydantic AI의 멀티모달 콘텐츠 타입 중 하나와 일치하는 모양을 출력에 넣음으로써 파일을 반환할 수 있어요. Pydantic AI는 실행이 계속되기 전에 그것을 그 타입으로 역직렬화해요(왕복에 쓰이는 것과 같은 ToolReturnContent union을 통해). 타입의 snake_case 필드 이름을 사용하세요 — 그것들은 Vercel-케이스 페이로드가 아닌 Pydantic AI 모델로 검증되므로 media_type은 역직렬화되지만 mediaType은 불투명 dict로 남아요. URL 모양은 media_type을 지닐 수 있어요. 없으면 어댑터가 URL에서 하나를 추론하고, 미디어 타입을 추론할 수 없는 URL은 파일 대신 일반 매핑으로 에이전트에 도달해요. 아래 모양 중 하나를 명시하는 출력은 그 파일 이며, 타입이 선언하지 않는 키는 매핑과 함께 버려져요 — 그러니 그대로 돌려받길 원하는 어떤 출력에서도 kind를 빼두세요. 세 가지 모양이 지원돼요:

• 인라인 바이트 -- BinaryContent 모양, { kind: 'binary', media_type: 'image/png', data: <bytes> }(이미지 미디어 타입은 BinaryImage가 됨). data 필드는 base64 문자열, 또는 JavaScript 프론트엔드가 Uint8Array나 Node Buffer를 먼저 인코딩하지 않고 JSON.stringify로 전달할 때 만드는 원시 바이트 모양({ "0": 137, "1": 80, ... } 또는 { "type": "Buffer", "data": [137, 80, ...] })을 받아들여요 — 모두 와이어 경계에서 바이트로 정규화되므로, 클라이언트 측 도구가 손으로 base64 인코딩하지 않고 바이너리 데이터를 반환할 수 있어요.
• 파일 URL -- { kind: 'image-url', url: 'https://example.com/chart.png' } 또는 { kind: 'document-url', url: 'https://example.com/report.pdf' } 같은 FileUrl 모양. 바이트를 인라인하는 것보다 종종 더 효율적이에요. 참조만 와이어를 건너고 프로바이더가 파일을 직접 가져오니까요 — 파일이 프론트엔드가 신뢰하는 URL에 이미 있을 때 좋은 선택이에요. URL은 그것의 스킴이 어댑터의 allowed_file_url_schemes allowlist(기본 http/https)를 통과할 때만 존중돼요. trust model 참고.
• 프로바이더 호스팅 파일 -- UploadedFile 모양, { kind: 'uploaded-file', file_id: 'file-123', provider_name: 'openai' }, 이미 프로바이더의 저장소에 업로드된 파일을 참조. 서버가 자체 자격 증명으로 프로바이더의 파일 API에 대해 해석하므로 allow_uploaded_files가 True일 때만 존중돼요.

Message metadata

VercelAIAdapter.dump_messages는 ModelRequest.metadata와 ModelResponse.metadata의 애플리케이션 키를 Vercel AI UIMessage.metadata에 쓰고, 예약된 pydantic_ai 키 아래에 메시지 timestamp를 저장해 왕복에서 살아남게 해요. VercelAIAdapter.load_messages은 그 애플리케이션 키와 timestamp를 돌아오는 길에 복원해요. 프레임워크 예약 __pydantic_ai__ 네임스페이스는 양방향으로 제외돼요.

load_messages는 또한 UIMessage.id를 그 예약 네임스페이스에 유지하고, dump_messages는 그것을 다시 id로 사용하므로, 브라우저가 보낸 히스토리가 브라우저가 할당한 id로 돌아와요. 각 ModelRequest 또는 ModelResponse는 id 하나를 지니므로, 연속된 UIMessage가 하나의 메시지로 병합될 때(시스템 메시지 다음에 사용자 메시지 같은 경우) 마지막 id만 유지돼요. 대신 id를 직접 고르려면 dump_messages에 generate_message_id를 전달하세요.

스트리밍할 때 timestamp는 또한 마지막 단계 후 Vercel AI message-metadata 청크로 방출되므로, AI SDK UI를 사용하는 프론트엔드가 어시스턴트 메시지와 함께 그것을 유지할 수 있어요. 요청 측 메시지에는 유사한 청크가 없어요 — 스트리밍된 청크에서만 히스토리를 재구축하는 프론트엔드는 어시스턴트 응답에만 timestamp를 보는 반면, dump_messages는 양쪽을 채워요.

UIMessage.metadata는 완전히 클라이언트 제어되므로, timestamp가 왕복되는 유일한 서버 소유 필드예요. __pydantic_ai__ 네임스페이스와 usage, model_name, provider_* 같은 필드는 의도적으로 제외돼요 — 그것을 덤프하면 인프라 세부사항이 누출될 수 있고, 복원하면 서버가 소유하는 값에 클라이언트 제출 히스토리를 신뢰하게 되니까요. 프레임워크와 프로바이더 상태는 신뢰된 서버 측 저장소에 보관하세요. 명시적 사용자 제어 옵트인 뒤의 왕복 확장은 issue #5174에서 추적돼요.

Trust model

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

Compaction

CompactionPart는 Vercel AI data part(data-compaction)를 통해 왕복되므로, useChat 같은 프론트엔드가 메시지 히스토리를 쥐고 있을 때 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를 참고하세요.

Tool Approval

Note

도구 승인에는 프론트엔드의 AI SDK UI v6 이상이 필요해요.

Pydantic AI는 AI SDK UI로 human-in-the-loop 도구 승인 워크플로우를 지원하며, 사용자가 도구 실행 전에 실행을 승인하거나 거부할 수 있어요. 승인이 필요한 도구 설정에 대한 자세한 내용은 deferred tool calls 문서를 참고하세요.

도구 승인 스트리밍을 활성화하려면 dispatch_request에 sdk_version=6을 전달하세요:

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

sdk_version=6일 때 어댑터는:

1. requires_approval=True인 도구가 호출될 때 tool-approval-request 청크를 방출
2. 후속 요청에서 승인 응답을 자동으로 추출
3. 거부된 도구에 대해 tool-output-denied 청크를 방출

프론트엔드에서 AI SDK UI의 useChat 훅이 승인 흐름을 처리해요. 사전 빌드된 승인 UI에는 AI Elements의 Confirmation 컴포넌트를 사용하거나, 훅의 addToolApprovalResponse 함수로 자체 것을 만들 수 있어요.

도구 승인 응답은 설계상 요청에서 신뢰돼요. useChat의 addToolApprovalResponse와 참조 Next.js 백엔드를 통한 프로토콜 왕복과 일치해요. 결정 자체는 실제 JSON 불리언이어야 해요. approved는 엄격하게 타입이 지정되므로, 어떤 대용품 — 0이나 "false"만큼 1이나 "true"도 — 결정으로 강제되지 않고 요청 검증에 실패해요. 애플리케이션이 요청이 아닌 서버 측 상태에 승인 결정을 묶어야 한다면, DeferredToolRequests를 가로채고 승인 ID를 서버 측에 영속한 다음, 재개할 때 명시적 deferred_tool_results를 전달하세요.

Tool input validation

tool-input-available은 에이전트가 도구의 스키마와 어떤 커스텀 args_validator에 대해 호출을 검증한 뒤에 방출되므로, 청크는 인자가 허용 가능한 것으로 알려진 후에만 발화해요. 청크의 input 필드는 모델이 방출한 원시 인자를 지녀요.

검증이 실패하면 어댑터는 tool-input-available 대신 tool-input-error를 방출해요. 청크는 같은 tool_call_id, tool_name, input(원시 인자)에 더해 모델에 다시 보내질 메시지에서 렌더링된 error_text 필드를 지녀요. 검증 실패가 재시도 가능할 때 에이전트는 호출을 재시도하고(도구의 retries 설정 적용) 각 시도에 대해 새 tool-input-(available|error)를 방출해요. 커스텀 validator가 ToolFailed를 발생시키면 실패는 종결적이며 재시도가 이어지지 않아요.

System prompts and instructions

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

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

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

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

vercel_ai_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.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, manage_system_prompt='client'
    )

더 알아보기 (Learn more)

• UI Event Streams 개요
• AG-UI 프로토콜