Pydantic AI 怎么为模型请求、工具调用和 MCP 连接分别设置超时 Pydantic AI 怎么为模型请求、工具调用和 MCP 连接分别设置超时【免费下载链接】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一次 agent run 里单次模型请求、单次工具调用和 MCP 服务器连接各自面临“最多能花多久”的问题。Pydantic AI 把这几类期限拆成了互不相干的配置项超一个不会连带超其他的各自到期后的表现也完全不同。这篇文章分别把这三处超时配起来单次模型请求的超时、函数工具的执行超时、MCP 连接与请求的超时并说明每处超时到期后到底发生什么、如何观察结果。前提Python 3.10已安装 Pydantic AIpip/uv-add pydantic-ai完整包自带 MCP 支持如果只用pydantic-ai-slim需要加mcp可选组。先对照每类超时要配哪个参数Timeouts 文档给出的总表里与本文目标直接相关的三行是要限制的单元配置方式到期后发生什么单次模型请求provider SDK 的重试会对每次尝试重新计时[ModelSettings][pydantic_ai.settings.ModelSettings] 上的timeoutprovider 客户端抛错除非FallbackModel或 transport retry 接管否则 run 失败一次函数工具调用Agent(tool_timeout...)或单个工具上的timeout模型收到重试提示Timed out after N seconds.消耗该工具的 retry 预算连接 MCP 服务器MCPToolset(init_timeout...)默认5秒连接与initialize握手失败单次 MCP 请求MCPToolset(read_timeout...)默认300秒请求失败在默认tool_error_behaviorretry下模型会把它当作可重试的工具错误注意这些配置项没有任何一项限制整次 run 的墙钟时长整 run 的期限是另一套机制见文末。为单次模型请求设置超时from pydantic_ai import Agent, ModelSettings agent Agent( test, # 替换为你的模型名字符串或模型实例 model_settingsModelSettings(timeout30), )ModelSettings.timeout的作用是 override 客户端级别的默认请求超时单位秒数字秒在所有支持的 provider 上都有效也接受 legacyhttpx.Timeout会被转换成 SDK 需要的类型。依据 settings.py 中的字段说明与 Timeouts 文档该设置按模型类生效不是全局生效。明确支持转发到 provider 客户端的模型类包括OpenAI、OpenAI Codex、Anthropic、Google、Groq、Mistral、Cerebras、Crusoe、GitHub Copilot、Ollama、OpenRouter、Snowflake、Z.AI、Bedrock Mantle其中Google 和 Mistral 只接受数字秒数不接受httpx.Timeout对象。基于 OpenAI 构建的模型类OpenAIChatModel/OpenAIResponsesModel继承这一转发能力。其他模型类会忽略这个设置转而适用构建时传入的 HTTP 客户端上的超时Pydantic AI 自己创建该客户端时默认是 600 秒总超时、5 秒连接超时。如果你的模型类忽略ModelSettings.timeout把超时配在 provider 真正接收超时的地方多数 provider 接受你传入的http_client但有几个例外——XaiProvider接受客户端级timeout或预配置的xai_clientBedrockProvider接受aws_read_timeout与aws_connect_timeout或预配置的bedrock_clientHuggingFaceProvider直接拒绝http_client改用hf_client。超时到期后的表现文档记载provider 客户端抛错run 失败除非有FallbackModel或 transport retry 接管这次失败。为函数工具设置执行超时import asyncio from pydantic_ai import Agent # Set a default timeout for the agents own tools agent Agent(test, tool_timeout30) agent.tool_plain async def slow_tool() - str: This tool will use the agents default timeout (30 seconds). await asyncio.sleep(10) return Done agent.tool_plain(timeout5) async def fast_tool() - str: This tool has its own timeout (5 seconds) that overrides the agent default. await asyncio.sleep(1) return Done以上是 Tool Timeout 文档中的示例两级配置的用法是Agent 级Agent(tool_timeout30)给注册在该 agent 上的工具一个默认超时。工具级通过agent.tool、agent.tool_plain或Tooldataclass 设置timeout覆盖 agent 级默认值。到期后的表现工具调用被当作可重试的失败处理模型收到重试提示Timed out after {timeout} seconds.并计入该工具的 retry 上限与参数校验错误或显式ModelRetry的处理方式一致。三条必须知道的边界来自 Timeouts 文档工具超时只由FunctionToolset强制且每个 toolset 各自携带一份。Agent(tool_timeout...)只覆盖注册在 agent 上的工具不会作用到你自己构造、经toolsets[...]传入的FunctionToolset。要给那个 toolset 单独设置FunctionToolset(timeout...)或在个别工具上设timeout。MCP 服务器提供的工具、外部 toolset、自定义AbstractToolset不读取这两项配置它们的期限要在服务器侧或传输层设置——也就是下一节的read_timeout。同步def工具不会被真正停止期限是在 await 外围强制的工作线程会一直跑到结束。另外如果你在工具函数内部自己实现期限、抛出TimeoutError应捕获后重新抛为ModelRetry或ToolFailed不要让它直接逃逸工具没配timeout时裸TimeoutError是普通异常会传播出agent.run()工具配了timeout时调用运行在anyio.fail_after(timeout)里你抛出的TimeoutError与期限到期无法区分会被包装成同一条Timed out after N seconds.重试提示——报告一个可能根本没到过的期限。验证方式运行 agent 并让某个工具超过其超时按文档记载模型收到的消息里会出现Timed out after {timeout} seconds.这条重试提示且该次重试计入工具 retry 预算retry 预算耗尽后 run 失败。为 MCP 连接和请求设置超时以 MCP Client 文档中的连接形式为例把两个超时参数传给MCPToolsetfrom pydantic_ai.mcp import MCPToolset toolset MCPToolset( http://localhost:8000/mcp, # 替换为你的 MCP 服务器地址stdio 服务器可改用 StdioTransport(...) init_timeout10, # 连接与 initialize 握手的超时秒 read_timeout60, # 单次 MCP 请求等待新消息的上限秒 )参数含义与默认值见 Timeouts 文档 与 MCPToolset 源码init_timeout初始连接与initialize握手的超时默认5 秒。到期时连接与握手失败。read_timeout在长连接上等待新消息的最长时间默认300 秒。到期时该次请求失败在默认tool_error_behaviorretry下见 MCP 文档的 Tool errors 一节模型会把它当作可重试的工具错误来适应而不是让整个 run 崩溃。需要提醒的是V1 的 MCP 客户端类合并为单个MCPToolset后init_timeout、read_timeout等的默认值与 V1 类不同从 V1 迁移过来时不能假设原超时配置仍然生效见 migration 文档。这些超时不覆盖的范围三个配置项各自限制的都是一“步”Timeouts 文档明确说明没有任何一项限制整次 run 的墙钟时长。如果你还需要下面两个期限用文档给出的对应机制整次 run 的时长没有内建开关用asyncio.timeoutPython 3.11或anyio.fail_after()包住agent.run()或从定时器里取消一个CancellationTokenrun 会按 Cancelling a Run 描述的方式被取消。run 的总工作量用UsageLimits限制请求数、工具调用数、token 数或花费到期抛UsageLimitExceeded。工具调用本身想“结束 run”而不是重试与超时是两回事抛出ModelRetry/ToolFailed会让 run 继续调用RunContext.cancel()会让 run 以RunCancelled结束见 工具高级文档。可进一步阅读Timeouts、Tool Timeout、MCP Client、Retries、ModelSettings API。【免费下载链接】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),仅供参考