使用 Pydantic AI 的 VercelAIAdapter 对接 Vercel AI Data Stream Protocol 构建流式聊天后端【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai导读本指南讲解 Pydantic AI 官方提供的 Vercel AI 集成能力通过VercelAIAdapter将前端AI SDK UI 的useChat等 hooks发来的 Vercel AI 请求输入转换为Agent.run_stream_events()的参数运行 Agent 后把 Pydantic AI 事件流实时转换为 Vercel AI 的 SSE 事件流返回前端。读完本文你将掌握 Starlette/FastAPI 下的一行式接入、无 Starlette 框架下的手动编排、取消与工具审批、自定义事件数据下发、客户端工具文件回传、消息元数据往返以及系统提示词与信任模型的完整配置。协议对接的总体架构Vercel AI Data Stream Protocol 是 AI SDK UI 生态useChat、useAssistant等 hooks 与 AI Elements 组件默认使用的流式协议。Pydantic AI 在pydantic_ai.ui.vercel_ai模块中实现了该协议的双向转换核心是两个类VercelAIAdapter负责把前端请求体转换为Agent.run_stream_events()的输入、运行 Agent、再把 Pydantic AI 事件转换为 Vercel AI 事件。它继承自抽象的UIAdapter基类与 AG-UI 协议 的AGUIAdapter平级二者共用同一套适配器骨架。VercelAIEventStream负责把 Pydantic AI 原生事件流转换为 Vercel AI 的 chunk 序列并编码为 SSE 字符串。常规请求路径下你无需直接使用它只有 Agent 事件不是经由服务于前端的那个请求到达你时例如 durable execution 工作流、消息队列或 WebSocket 扇出场景才需要单独实例化它来转换事件参见 UI Event Streams 文档 中的encode_events示例。从源码看VercelAIAdapter是一个 dataclass声明了sdk_version默认5、server_message_id等字段并复用了UIAdapter提供的dispatch_request、from_request、run_stream、run_stream_native、encode_stream、streaming_response、transform_stream等完整方法族。整个请求处理链条为build_run_input(请求体)→VercelAIAdapter(agent, run_input, accept)→run_stream(...)→encode_stream(...)→StreamingResponse。与 Starlette / FastAPI 一行式接入如果后端基于 Starlette 系框架FastAPI、Litestar 等VercelAIAdapter.dispatch_request()类方法可以直接在端点函数中消费请求并返回流式响应这是官方推荐的最简路径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, agentagent)除了request与agentdispatch_request还接受与Agent.run_stream_events()相同的可选参数message_history、deferred_tool_results、conversation_id、run_id、model、instructions、deps、output_type、model_settings、usage_limits、usage、metadata、infer_name、toolsets、capabilities等以及两个回调on_completeAgent 运行成功时触发接收AgentRunResult可额外产出 Vercel AI 事件on_cancelAgent 因一等取消first-party cancellation结束时触发接收RunCancelled同样可额外产出事件。调用链在 UIAdapter.dispatch_request 中实现先from_request()解析请求若请求体校验失败则返回 422UNPROCESSABLE_ENTITYJSON 错误随后streaming_response(adapter.run_stream(...))生成流式响应。注意请求的Accept头会被用作流式响应的media_type。无 Starlette 框架时的手动编排Advanced Usage对于 Django、Flask 等非 Starlette 框架或需要细粒度控制输入输出的场景可以直接实例化VercelAIAdapter并链式调用其方法效果等价于dispatch_request解析请求体VercelAIAdapter.build_run_input(await request.body())返回 Vercel AI 的RequestData对象。其实现是request_data_ta.validate_json(body)见 _adapter.py即用 Pydantic TypeAdapter 严格校验 JSON。如果是 Starlette/FastAPI 环境也可直接用VercelAIAdapter.from_request()一步构建适配器实例。运行 Agentadapter.run_stream(...)运行 Agent 并返回 Vercel AI 事件流支持与Agent.run_stream_events()相同的参数及on_complete/on_cancel。它内部先走run_stream_native()返回 Pydantic AI 原生事件再经transform_stream()转换你也可以自己先拿run_stream_native()的原生事件流再调用transform_stream()手动转换。编码 SSEadapter.encode_stream(event_stream)把 Vercel AI 事件流编码为 SSE 字符串或者直接用adapter.streaming_response(...)生成 Starlette/FastAPI 的流式响应对象。完整的可运行示例含输入校验、422 返回、取消 token 注册表清理参见官方文档 run_stream.py 中的对应示例与下方取消一节。该示例使用 FastAPI但可改造适配任意 Web 框架。取消机制一等取消与外部取消当一个运行以一等取消结束来自工具内ctx.cancel()、AgentRun.cancel()、或你服务器端在取消端点接线的CancellationToken适配器会向流中发射一个 Vercelabortchunk。useChat会保留已产生的部分消息并在onFinish中上报isAbort而不是进入错误状态。此时on_cancel回调被触发可在其中持久化可恢复的消息历史RunCancelled.all_messages()返回可继续传递的消息列表。重要区别客户端调用stop()会直接中断浏览器请求服务器侧看到的是一次断开disconnect属于外部取消——运行以asyncio.CancelledError方式被撕裂不会发射abortchunkon_cancel也不会触发客户端反正已经断开了。若希望在用户点停止时也能拿到abortchunk 并执行on_cancel应当保持流连接改用一等取消给运行传入CancellationToken并暴露一个独立端点如POST /chat/{id}/cancel调用token.cancel()。完整实现如下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(fcancelled 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( contentjson.dumps(e.json()), media_typeapplication/json, status_codeHTTPStatus.UNPROCESSABLE_ENTITY, ) adapter VercelAIAdapter(agentagent, run_inputrun_input, acceptaccept) cancellation_token CancellationToken() cancellation_tokens[chat_id] cancellation_token event_stream adapter.run_stream( cancellation_tokencancellation_token, on_cancelon_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_typeaccept) app.post(/chat/{chat_id}/cancel, status_codeHTTPStatus.NO_CONTENT) async def cancel_chat(chat_id: str) - None: if token : cancellation_tokens.get(chat_id): token.cancel()这是需要持久化的可恢复历史——在后续运行中作为message_history传入即可续接对话。注意上述内存 token 注册表要求单进程部署或粘性路由。多 worker 部署时需借助消息代理等共享协调机制把取消请求路由到持有该运行的 worker。从源码看VercelAIEventStream.on_cancelled()发射的AbortChunk内容为reasonThe agent run was cancelled.见 _event_stream.py而on_error()会设置finish_reasonerror并发射ErrorChunk_event_stream.py。向客户端下发数据Data Chunks运行过程中例如长耗时工具的执行进度想向客户端推送自定义数据可以在工具内通过ctx.emit()发射一个CustomEventfrom dataclasses import dataclass from pydantic_ai import Agent, CustomEvent, RunContext agent Agent(openai:gpt-5.2) dataclass(kw_onlyTrue) 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): # 完成一个单位的工作然后告诉前端进度 await ctx.emit(FileUploadProgressEvent(donedone, totaltotal)) return fUploaded {total} files每个事件会以DataChunk形式到达客户端type为data-{name}data为to_payload()的结果——上例即typedata-file_upload_progress、data{done: 1, total: 3}。chunk 在事件发射的当下、工具仍在运行时即到达天然支持进度条。形状一致性无论事件是否从工具调用内部发射data形状都一致因此前端针对某一形状编写的代码不会因为同一事件类日后改在别处发射而失效。如需自定义形状比如按前端期望命名字段、把工具归属信息放到线上覆盖to_payload()即可from dataclasses import dataclass from typing import Any from pydantic_ai import CustomEvent dataclass(kw_onlyTrue) 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()返回一个数据承载型 chunk见下文则该 chunk 原样透传事件类声明为uiFalse时永不下发仅服务端消费的事件留在服务端进程从未 import 过的事件类也不会被转发因为其 opt-out 标记是挂在类上而非走线上。工具返回时携带 chunk工具可通过返回带metadata单个或列表的ToolReturn对象把 Vercel AI data stream chunk 附加到工具结果上。支持四种类型DataChunk、SourceUrlChunk、SourceDocumentChunk、FileChunkfrom 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_valuefFound 2 results for {query}, metadata[ SourceUrlChunk( source_iddoc-1, urlhttps://example.com/docs/intro, titleIntroduction, ), DataChunk( typedata-search-results, data{query: query, count: 2}, ), ], )与ctx.emit发射的事件不同这些 chunk 属于消息的一部分能随消息历史往返而幸存——这正是前端需要重建的数据如答案背后的来源 URL所期望的代价是它们在工具返回时而非运行过程中下发。源码中iter_metadata_chunks()见 _utils.py只会转发上述四种>app.post(/chat) async def chat(request: Request) - Response: return await VercelAIAdapter.dispatch_request(request, agentagent, sdk_version6)当sdk_version6时适配器会在调用requires_approvalTrue的工具时发射tool-approval-requestchunk自动从后续请求中提取审批响应为被拒绝的工具发射tool-output-deniedchunk。前端方面AI SDK UI 的useChathook 处理审批流程可使用 AI Elements 的Confirmation组件做现成审批 UI或用 hook 的addToolApprovalResponse自行构建。审批响应的严格性审批响应按协议经useChat的addToolApprovalResponse与参考 Next.js 后端往返设计上被信任。但审批决定本身必须是真正的 JSON 布尔值——ToolApprovalResponded.approved字段是严格布尔StrictBool任何替代值1、true、0、false都会导致请求校验失败而不会被强制转换成一个决定见 request_types.py 的注释防止{approved: 1}被宽松模式强转为通过。若需要把审批决定绑定到服务端状态而非请求本身可拦截DeferredToolRequests在服务端持久化审批 ID并在续接时显式传入deferred_tool_results。补充说明源码中sdk_version类型为Literal[5, 6, 7]默认5以保证向后兼容7的线协议与6完全相同v7 的>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, agentagent, manage_system_promptclient )协议细节的源码级补充Finish reason 映射Pydantic AI 的结束原因经_FINISH_REASON_MAP_event_stream.py映射为 Vercel 格式stop→stop、length→length、content_filter→content-filter、tool_call→tool-calls、error→error未知值统一为other。SSE 编码与响应头每个 chunk 编码为data: {json}\n\nencode_event且流式响应带x-vercel-ai-ui-message-stream: v1响应头VERCEL_AI_DSP_HEADERS这是 AI SDK UI 识别 data stream 协议的标准信号。消息 ID 生成_generate_message_id按优先级生成确定性消息 ID有provider_response_id用{provider_response_id}-{index}有run_id用{run_id}-{index}否则用uuid5(timestamp-kind-role-index)_adapter.py。conversation_id关联适配器把请求体顶层的idchat ID作为conversation_id用于跨多次运行关联 OpenTelemetry span 与消息历史。往返的已知损耗dump_messages → load_messages对工具结果并非完全无损——RetryPromptPart重新加载后会变成outcomefailed的ToolReturnPart协议没有独立 retry 概念无法解析为 JSON 对象的ToolCallPart.args会被重写为{INVALID_JSON: raw args}。需要 retry 语义跨往返存活时应在进程内维护对话而非经由 Vercel AI 线格式持久化。验证与测试仓库的测试套件tests/test_ui.py覆盖了适配器的请求解析、事件流编码与消息往返tests/test_vercel_ai.py与tests/cassettes/test_vercel_ai/中的 VCR 磁带用于验证真实模型调用下的流式行为。测试同时覆盖了sdk_version5/6两种线格式下的 chunk 输出差异可作为理解协议行为的补充参考。总结VercelAIAdapter把 Pydantic AI 的 Agent 运行无缝接入 Vercel AI Data Stream Protocol 生态dispatch_request一行式接入 FastAPI手动编排适配任意框架取消、自定义数据、工具审批、消息元数据与压缩均开箱即用。安全方面系统提示词归属、文件 URL scheme 白名单与上传文件开关共同构成默认信任模型在对外暴露该端点时请务必把它当作内部后端服务放在你自己已认证的路由处理器之内。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 Pydantic AI 的 VercelAIAdapter 对接 Vercel AI Data Stream Protocol 构建流式聊天后端
使用 Pydantic AI 的 VercelAIAdapter 对接 Vercel AI Data Stream Protocol 构建流式聊天后端【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai导读本指南讲解 Pydantic AI 官方提供的 Vercel AI 集成能力通过VercelAIAdapter将前端AI SDK UI 的useChat等 hooks发来的 Vercel AI 请求输入转换为Agent.run_stream_events()的参数运行 Agent 后把 Pydantic AI 事件流实时转换为 Vercel AI 的 SSE 事件流返回前端。读完本文你将掌握 Starlette/FastAPI 下的一行式接入、无 Starlette 框架下的手动编排、取消与工具审批、自定义事件数据下发、客户端工具文件回传、消息元数据往返以及系统提示词与信任模型的完整配置。协议对接的总体架构Vercel AI Data Stream Protocol 是 AI SDK UI 生态useChat、useAssistant等 hooks 与 AI Elements 组件默认使用的流式协议。Pydantic AI 在pydantic_ai.ui.vercel_ai模块中实现了该协议的双向转换核心是两个类VercelAIAdapter负责把前端请求体转换为Agent.run_stream_events()的输入、运行 Agent、再把 Pydantic AI 事件转换为 Vercel AI 事件。它继承自抽象的UIAdapter基类与 AG-UI 协议 的AGUIAdapter平级二者共用同一套适配器骨架。VercelAIEventStream负责把 Pydantic AI 原生事件流转换为 Vercel AI 的 chunk 序列并编码为 SSE 字符串。常规请求路径下你无需直接使用它只有 Agent 事件不是经由服务于前端的那个请求到达你时例如 durable execution 工作流、消息队列或 WebSocket 扇出场景才需要单独实例化它来转换事件参见 UI Event Streams 文档 中的encode_events示例。从源码看VercelAIAdapter是一个 dataclass声明了sdk_version默认5、server_message_id等字段并复用了UIAdapter提供的dispatch_request、from_request、run_stream、run_stream_native、encode_stream、streaming_response、transform_stream等完整方法族。整个请求处理链条为build_run_input(请求体)→VercelAIAdapter(agent, run_input, accept)→run_stream(...)→encode_stream(...)→StreamingResponse。与 Starlette / FastAPI 一行式接入如果后端基于 Starlette 系框架FastAPI、Litestar 等VercelAIAdapter.dispatch_request()类方法可以直接在端点函数中消费请求并返回流式响应这是官方推荐的最简路径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, agentagent)除了request与agentdispatch_request还接受与Agent.run_stream_events()相同的可选参数message_history、deferred_tool_results、conversation_id、run_id、model、instructions、deps、output_type、model_settings、usage_limits、usage、metadata、infer_name、toolsets、capabilities等以及两个回调on_completeAgent 运行成功时触发接收AgentRunResult可额外产出 Vercel AI 事件on_cancelAgent 因一等取消first-party cancellation结束时触发接收RunCancelled同样可额外产出事件。调用链在 UIAdapter.dispatch_request 中实现先from_request()解析请求若请求体校验失败则返回 422UNPROCESSABLE_ENTITYJSON 错误随后streaming_response(adapter.run_stream(...))生成流式响应。注意请求的Accept头会被用作流式响应的media_type。无 Starlette 框架时的手动编排Advanced Usage对于 Django、Flask 等非 Starlette 框架或需要细粒度控制输入输出的场景可以直接实例化VercelAIAdapter并链式调用其方法效果等价于dispatch_request解析请求体VercelAIAdapter.build_run_input(await request.body())返回 Vercel AI 的RequestData对象。其实现是request_data_ta.validate_json(body)见 _adapter.py即用 Pydantic TypeAdapter 严格校验 JSON。如果是 Starlette/FastAPI 环境也可直接用VercelAIAdapter.from_request()一步构建适配器实例。运行 Agentadapter.run_stream(...)运行 Agent 并返回 Vercel AI 事件流支持与Agent.run_stream_events()相同的参数及on_complete/on_cancel。它内部先走run_stream_native()返回 Pydantic AI 原生事件再经transform_stream()转换你也可以自己先拿run_stream_native()的原生事件流再调用transform_stream()手动转换。编码 SSEadapter.encode_stream(event_stream)把 Vercel AI 事件流编码为 SSE 字符串或者直接用adapter.streaming_response(...)生成 Starlette/FastAPI 的流式响应对象。完整的可运行示例含输入校验、422 返回、取消 token 注册表清理参见官方文档 run_stream.py 中的对应示例与下方取消一节。该示例使用 FastAPI但可改造适配任意 Web 框架。取消机制一等取消与外部取消当一个运行以一等取消结束来自工具内ctx.cancel()、AgentRun.cancel()、或你服务器端在取消端点接线的CancellationToken适配器会向流中发射一个 Vercelabortchunk。useChat会保留已产生的部分消息并在onFinish中上报isAbort而不是进入错误状态。此时on_cancel回调被触发可在其中持久化可恢复的消息历史RunCancelled.all_messages()返回可继续传递的消息列表。重要区别客户端调用stop()会直接中断浏览器请求服务器侧看到的是一次断开disconnect属于外部取消——运行以asyncio.CancelledError方式被撕裂不会发射abortchunkon_cancel也不会触发客户端反正已经断开了。若希望在用户点停止时也能拿到abortchunk 并执行on_cancel应当保持流连接改用一等取消给运行传入CancellationToken并暴露一个独立端点如POST /chat/{id}/cancel调用token.cancel()。完整实现如下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(fcancelled 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( contentjson.dumps(e.json()), media_typeapplication/json, status_codeHTTPStatus.UNPROCESSABLE_ENTITY, ) adapter VercelAIAdapter(agentagent, run_inputrun_input, acceptaccept) cancellation_token CancellationToken() cancellation_tokens[chat_id] cancellation_token event_stream adapter.run_stream( cancellation_tokencancellation_token, on_cancelon_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_typeaccept) app.post(/chat/{chat_id}/cancel, status_codeHTTPStatus.NO_CONTENT) async def cancel_chat(chat_id: str) - None: if token : cancellation_tokens.get(chat_id): token.cancel()这是需要持久化的可恢复历史——在后续运行中作为message_history传入即可续接对话。注意上述内存 token 注册表要求单进程部署或粘性路由。多 worker 部署时需借助消息代理等共享协调机制把取消请求路由到持有该运行的 worker。从源码看VercelAIEventStream.on_cancelled()发射的AbortChunk内容为reasonThe agent run was cancelled.见 _event_stream.py而on_error()会设置finish_reasonerror并发射ErrorChunk_event_stream.py。向客户端下发数据Data Chunks运行过程中例如长耗时工具的执行进度想向客户端推送自定义数据可以在工具内通过ctx.emit()发射一个CustomEventfrom dataclasses import dataclass from pydantic_ai import Agent, CustomEvent, RunContext agent Agent(openai:gpt-5.2) dataclass(kw_onlyTrue) 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): # 完成一个单位的工作然后告诉前端进度 await ctx.emit(FileUploadProgressEvent(donedone, totaltotal)) return fUploaded {total} files每个事件会以DataChunk形式到达客户端type为data-{name}data为to_payload()的结果——上例即typedata-file_upload_progress、data{done: 1, total: 3}。chunk 在事件发射的当下、工具仍在运行时即到达天然支持进度条。形状一致性无论事件是否从工具调用内部发射data形状都一致因此前端针对某一形状编写的代码不会因为同一事件类日后改在别处发射而失效。如需自定义形状比如按前端期望命名字段、把工具归属信息放到线上覆盖to_payload()即可from dataclasses import dataclass from typing import Any from pydantic_ai import CustomEvent dataclass(kw_onlyTrue) 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()返回一个数据承载型 chunk见下文则该 chunk 原样透传事件类声明为uiFalse时永不下发仅服务端消费的事件留在服务端进程从未 import 过的事件类也不会被转发因为其 opt-out 标记是挂在类上而非走线上。工具返回时携带 chunk工具可通过返回带metadata单个或列表的ToolReturn对象把 Vercel AI data stream chunk 附加到工具结果上。支持四种类型DataChunk、SourceUrlChunk、SourceDocumentChunk、FileChunkfrom 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_valuefFound 2 results for {query}, metadata[ SourceUrlChunk( source_iddoc-1, urlhttps://example.com/docs/intro, titleIntroduction, ), DataChunk( typedata-search-results, data{query: query, count: 2}, ), ], )与ctx.emit发射的事件不同这些 chunk 属于消息的一部分能随消息历史往返而幸存——这正是前端需要重建的数据如答案背后的来源 URL所期望的代价是它们在工具返回时而非运行过程中下发。源码中iter_metadata_chunks()见 _utils.py只会转发上述四种>app.post(/chat) async def chat(request: Request) - Response: return await VercelAIAdapter.dispatch_request(request, agentagent, sdk_version6)当sdk_version6时适配器会在调用requires_approvalTrue的工具时发射tool-approval-requestchunk自动从后续请求中提取审批响应为被拒绝的工具发射tool-output-deniedchunk。前端方面AI SDK UI 的useChathook 处理审批流程可使用 AI Elements 的Confirmation组件做现成审批 UI或用 hook 的addToolApprovalResponse自行构建。审批响应的严格性审批响应按协议经useChat的addToolApprovalResponse与参考 Next.js 后端往返设计上被信任。但审批决定本身必须是真正的 JSON 布尔值——ToolApprovalResponded.approved字段是严格布尔StrictBool任何替代值1、true、0、false都会导致请求校验失败而不会被强制转换成一个决定见 request_types.py 的注释防止{approved: 1}被宽松模式强转为通过。若需要把审批决定绑定到服务端状态而非请求本身可拦截DeferredToolRequests在服务端持久化审批 ID并在续接时显式传入deferred_tool_results。补充说明源码中sdk_version类型为Literal[5, 6, 7]默认5以保证向后兼容7的线协议与6完全相同v7 的>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, agentagent, manage_system_promptclient )协议细节的源码级补充Finish reason 映射Pydantic AI 的结束原因经_FINISH_REASON_MAP_event_stream.py映射为 Vercel 格式stop→stop、length→length、content_filter→content-filter、tool_call→tool-calls、error→error未知值统一为other。SSE 编码与响应头每个 chunk 编码为data: {json}\n\nencode_event且流式响应带x-vercel-ai-ui-message-stream: v1响应头VERCEL_AI_DSP_HEADERS这是 AI SDK UI 识别 data stream 协议的标准信号。消息 ID 生成_generate_message_id按优先级生成确定性消息 ID有provider_response_id用{provider_response_id}-{index}有run_id用{run_id}-{index}否则用uuid5(timestamp-kind-role-index)_adapter.py。conversation_id关联适配器把请求体顶层的idchat ID作为conversation_id用于跨多次运行关联 OpenTelemetry span 与消息历史。往返的已知损耗dump_messages → load_messages对工具结果并非完全无损——RetryPromptPart重新加载后会变成outcomefailed的ToolReturnPart协议没有独立 retry 概念无法解析为 JSON 对象的ToolCallPart.args会被重写为{INVALID_JSON: raw args}。需要 retry 语义跨往返存活时应在进程内维护对话而非经由 Vercel AI 线格式持久化。验证与测试仓库的测试套件tests/test_ui.py覆盖了适配器的请求解析、事件流编码与消息往返tests/test_vercel_ai.py与tests/cassettes/test_vercel_ai/中的 VCR 磁带用于验证真实模型调用下的流式行为。测试同时覆盖了sdk_version5/6两种线格式下的 chunk 输出差异可作为理解协议行为的补充参考。总结VercelAIAdapter把 Pydantic AI 的 Agent 运行无缝接入 Vercel AI Data Stream Protocol 生态dispatch_request一行式接入 FastAPI手动编排适配任意框架取消、自定义数据、工具审批、消息元数据与压缩均开箱即用。安全方面系统提示词归属、文件 URL scheme 白名单与上传文件开关共同构成默认信任模型在对外暴露该端点时请务必把它当作内部后端服务放在你自己已认证的路由处理器之内。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考