pydantic-ai

An AI agent failing because one model stops responding is a product problem, not AI magic. Pydantic AI added ModelFallback and a timeout for streams that never start. The boring reliability work is what makes agents usable. github.com/pydantic/pydantic-ai/releases/tag/v2.6…

便利なAIが増えるほど、問われるのは何でも覚えている能力だけではありません。 人間が止めたことを、きちんと止められるか。 その境界を守ることも、AIを社会で使うための重要な条件です。 #ぶるーすかい？ 【GitHub一次資料】 不具合の報告・再現 pydantic-ai#11264 私の指摘 pydantic-ai#11264 メンテナによる修正・テスト・マージ pydantic-ai#11394

Pydantic AI's MCP telemetry could leak auth tokens into OpenTelemetry spans even with `include_content=False`. Fixed in 2.58.0 / 1.107.8. Old traces still need access review and credential rotation. github.com/pydantic/pydantic-ai/security/advisori…

include_content=False Didn't Stop pydantic-ai's MCP Token Leak Originally published at webofmike.com on 2026-10-26. The demo repo and every command in it were run before publishing. pydantic-ai shipped two security fixes on 2026-10-19. The one I reproduced, GHSA-xv4x-g8mx-ch56, is a two-flag problem: include_content=False does exactly what it says for prompts and tool calls, and a second, separate flag still wrote MCP server URLs, authorization tokens, and custom header values into an OpenTelemetry span in plaintext. I built themsquared/pydantic-ai-mcp-telemetry-leak to see the exact span attribute that leaked, confirm the fix closes it, and confirm the thing include_content=False was actually supposed to cover still works. What the advisory covers pydantic-ai lets an agent reach an MCP server two ways. MCPToolset is pydantic-ai's own client: it connects to the server directly, and the URL and headers never leave your process. MCPServerTool is provider-native: you hand the server's URL, an authorization_token, and headers to the model provider (Anthropic, OpenAI Responses, xAI), and the provider connects to the MCP server on your agent's behalf. Anthropic's beta MCP connector and OpenAI Responses' remote MCP tool both work this way. That provider-native config has to travel somewhere: it's part of the request pydantic-ai builds for the provider's API. pydantic-ai's own instrumentation layer records that request shape in an OpenTelemetry span, under the attribute model_request_parameters, so you can see what tools and settings a call used. Before 2.58.0, that attribute included the full MCPServerTool dataclass, authorization_token and headers included, regardless of include_content. Reproducing it without a network call The leak lives in pydantic-ai's instrumentation code (_instrumentation.py), which wraps every model provider identically. That means I didn't need a real Anthropic account, a real MCP server, or even network access to reproduce it: any Model subclass that declares it supports MCPServerTool exercises the same span-building path. class StubModel(Model): """Smallest possible Model: returns a canned response, no network, no API key.""" async def request(self, messages, model_settings, model_request_parameters): return ModelResponse( parts=[TextPart(content="ok")], usage=RequestUsage(input_tokens=1, output_tokens=1), model_name="stub", timestamp=datetime.now(timezone.utc), ) @classmethod def supported_native_tools(cls): return frozenset({MCPServerTool}) Wrap it in InstrumentedModel with include_content=False, the setting someone reaches for specifically to keep prompts and tool arguments out of traces, and hand it a native MCP tool config carrying a fake token: mcp_tool = MCPServerTool( id="internal-billing-mcp", url="https://internal-mcp.example.com/mcp", authorization_token="sk-mcp-prod-SUPER-SECRET-TOKEN-abc123", headers={"X-Tenant-Key": "tenant-9f8e-SECRET"}, ) settings = InstrumentationSettings(include_content=False) model = InstrumentedModel(StubModel(), settings) params = ModelRequestParameters(native_tools=[mcp_tool]) await model.request(messages, None, params) Capture the span with an in-memory OpenTelemetry exporter and read model_request_parameters back off it. What came back Against pydantic-ai-slim==2.56.0, with include_content=False set on this exact span: {"native_tools":[{"kind":"mcp_server","id":"internal-billing-mcp", "url":"https://internal-mcp.example.com/mcp", "authorization_token":"sk-mcp-prod-SUPER-SECRET-TOKEN-abc123", "headers":{"X-Tenant-Key":"tenant-9f8e-SECRET"}}], ...} The token and the custom header are both right there, unredacted, on a span whose whole point was to not carry sensitive content. Meanwhile, on the same span, gen_ai.input.messages correctly came back as structure only, [{"role":"user","parts":[{"type":"text"}]}], no actual prompt text. include_content=False worked exactly as documented for the thing it's documented to cover. It just doesn't cover model_request_parameters, which ships under its own flag, include_model_request_parameters, defaulting to True. Someone configuring include_content=False to keep secrets out of their tracing backend had no reason to know a second flag existed, let alone that it defaulted the other way. Same script, same inputs, against pydantic-ai-slim==2.58.0: {"native_tools":[{"kind":"mcp_server","id":"internal-billing-mcp", "url":"[REDACTED]","authorization_token":"[REDACTED]", "headers":{"X-Tenant-Key":"[REDACTED]"}}], ...} The fix redacts the specific fields that carry credentials inside a native tool's config, independent of either flag. model_request_parameters still tells you a native MCP tool was in play; it no longer tells you what it would take to use it. Why this is worse than a logging bug Traces usually land somewhere with a different access model than your secret store: a shared observability platform, a third-party APM vendor, a teammate who has trace-viewer access but not secrets-manager access. None of those audiences is supposed to be able to read your MCP server's bearer token. Before 2.58.0, every traced call through a provider-native MCP connector handed it to all of them, in plaintext, in a system most people don't think of as a place secrets end up. If you're on an affected version (pydantic-ai-slim before 2.58.0, or before 1.107.8 on the 1.x line): upgrade, restrict access to historical traces in the meantime, and rotate any credential that ever passed through a provider-native MCP tool config while instrumented. The fix is a point upgrade, not a config change, which is exactly why include_content=False alone wasn't going to save you. The same release also patched GHSA-5pfm-2g3g-gh85: with non-empty server-side message_history and the default manage_system_prompt='server', Vercel AI and AG-UI adapters could execute an available tool with client-chosen arguments without an actual model request. I didn't reproduce that one; it needs a live UI adapter session rather than a single instrumented span, which is a different repo. It's the same release and the same lesson from a different angle: an agent framework's convenience defaults are worth re-reading every time they patch one. agentevals' OTel-native scoring ships eval results through the same kind of pipeline this post is about: an agent's run reduced to spans in a trace backend. That's exactly the surface worth auditing for what else ends up in those spans besides the thing you meant to measure. Try it git clone https://github.com/themsquared/pydantic-ai-mcp-telemetry-leak.git cd pydantic-ai-mcp-telemetry-leak ./scripts/demo.sh One script, two throwaway virtualenvs, no API key. If you're instrumenting any agent framework with OpenTelemetry and relying on a single "don't log content" flag, this is worth fifteen minutes: check whether your framework ships a second flag for tool and connector configuration, and whether it defaults the way you'd assume. Repo: themsquared/pydantic-ai-mcp-telemetry-leak. Frequently asked questions What is GHSA-xv4x-g8mx-ch56? A Medium-severity advisory in pydantic-ai: a provider-native MCP connection's server URL, authorization token, and custom header values could land in plaintext in an OpenTelemetry request-span attribute, even with include_content=False set. It affects pydantic-ai-slim before 2.58.0 on the 2.x line and before 1.107.8 on the 1.x line, and is fixed in both. Does include_content=False stop pydantic-ai from logging MCP credentials in OpenTelemetry traces? Not before 2.58.0. include_content=False correctly withholds prompt and tool-call content, but a provider-native MCP tool's config is recorded in a separate model_request_parameters span attribute, governed by its own flag (include_model_request_parameters, default True). Setting include_content=False alone left that attribute, and any credentials in it, untouched. What is provider-native MCP in pydantic-ai, and how is it different from MCPToolset? MCPToolset is pydantic-ai's own client: it connects to an MCP server directly and the server's URL and headers never leave your process. A provider-native MCPServerTool instead hands the server's URL, an authorization_token, and headers to the model provider (Anthropic, OpenAI Responses, xAI), which connects to the MCP server itself. That config travels as part of the request pydantic-ai builds, which is what pydantic-ai's own instrumentation layer then recorded in the trace span. How can I reproduce the pydantic-ai MCP telemetry leak myself? themsquared/pydantic-ai-mcp-telemetry-leak runs one script against two virtualenvs, pydantic-ai-slim 2.56.0 and 2.58.0, with no network calls and no API key. A 15-line stub Model declares support for MCPServerTool and returns a canned response; an in-memory OpenTelemetry exporter captures the resulting span so you can diff the model_request_parameters attribute between versions directly. --- *Canonical version, with machine-readable markdown at https://webofmike.com/pydantic-ai-mcp-telemetry-leak/index.md: webofmike.com

Airflow × AI Pluginを試す 概要 いつの間にかAirflowでLLM・AI Agent向けのProviderがサポートされていたので、ClaudeCode × Airflowをいくつか試してみる。 事前準備 Airflow3.1をQuick Start通りにインストールし、standaloneで起動できるようにしておく。 standaloneにおいての実行で、forkされた子プロセスからHTTPS(claude)を呼ぶと不安定になるので、DAGの実行にはairflow dags test <dag名>を用いる。 Claude ConsoleでAPIを発行しておく。 試したことないが、ローカルLLMもサポートされている。(link) インストール provideインストール pip install 'apache-airflow-providers-common-ai[anthropic]' airflow connectionの追加 airflow connections add my_claude_conn \ --conn-type pydanticai \ --conn-password 'sk-ant-xxxxxxxx' \ --conn-extra '{"model": "anthropic:claude-haiku-5-5"}' localhost 試してみる @task.llm / LLMOperator airflow.apache.org 無難なプロンプトを投げるようなDecorator / Operator。 pydantic_aiにおいて、prompt返答の型安全性を実現し、pydantic_aiからauthropic-sdkを呼び出しClaudeへ接続している。 各種パラメータについて promptとsystem_promptがある。system_promptがagentに対するpromptであり、promptが具体的なタスクのprompt。 airflow.apache.org agent = self._create_agent(output_type=self.output_type, instructions=self.system_prompt) result = self._run_llm(context, agent, self.prompt, usage_limits=usage_limits) usage_limitsにおいて、tokenのlimitなどを設けることができる。 require_approvalにおいて、人手による承認を設けることができる。 実装例.おすすめの本を紹介してもらう データエンジニアとしてのおすすめの本を紹介してもらう実装を試す。 dag.py from datetime import timedelta from pydantic import BaseModel from airflow.providers.common.compat.sdk import dag, task @dag def my_pipeline(): class Book(BaseModel): title_english: str title_ja: str price: str class Books(BaseModel): books: list[Book] @task.llm( llm_conn_id="my_claude_conn", system_prompt="あなたはデータエンジニアのスペシャリストです", output_type=Books, allow_modifications=True, ) def recommend(topic: str): 以下のreturnがtask.llmに渡すpromptになる return f"{topic}として読むべき本を3冊おすすめしてください。" @task def show(result: dict): print("=== おすすめの本 ===") for i, book in enumerate(result["books"], start=1): print(f"{i}. {book['title_ja']}") print(f" 原題: {book['title_english']}") print(f" 価格: {book['price']}") show(recommend("データエンジニア")) my_pipeline() 実行した結果が以下です。@task.llmのoutput_typeに合った形(Books)でresult["books"]が返ってきていることがわかります。 === おすすめの本 === 1. データ指向アプリケーションデザイン 原題: Designing Data-Intensive Applications 価格: 約5,720円（税込） 2. データエンジニアリングの基礎 原題: Fundamentals of Data Engineering 価格: 約4,400円（税込） 3. データウェアハウス工具箱 原題: The Data Warehouse Toolkit 価格: 約6,600円（税込） 試しにrecommend関数に指定したpromptをoutput_typeに沿っていない形にして実行してみる。 ・・・ def recommend(topic: str): return "今日の天気は？" すると以下のようにエラーになる。あまりエラーが判りやすくはないですね。 2026-10-25T04:20:54.338091Z [error ] [DAG TEST] Error running task <TaskInstance: my_pipeline.recommend manual__2026-10-25T04:20:50.026474+00:00 [queued]> [airflow.sdk.definitions.dag] loc=dag.py:1419 Traceback (most recent call last): ・・・ File "/Users/srn221b/airflow/airflow_venv/lib/python3.12/site-packages/airflow/sdk/api/client.py", line 171, in get_json_error raise err airflow.sdk.api.client.ServerResponseError: Server returned error @task.agent / AgentOperator airflow.apache.org toolやskillなどを用いてLLMに考えてもらいながらtaskを実行するときに使えるDecorator / Operator。 claudeへの接続の仕方などは@task.llmと同様。 各種パラメータについて toolsetsにおいて、Database(SQLToolset)・AgentSkill（AgentSkillsToolset）など、LLMに答えてもらうにあたり利用したいtoolを設定できる。 capabilitiesにおいて、Thinking、WebSerachなどの能力を指定できる。 前回の結果を再利用する手段として3つがサポートされている。 durable cache_prompt message_history 目的 失敗したときのやり直しを安く、確実にする 同じ入力の送り直しを安くする 会話を次の実行に引き継ぐ 再利用する範囲 同じタスクインスタンスのリトライ間 1回の実行の中で繰り返すリクエスト間 別のタスク実行（別の DAG run など）の間 どこに保存されるか Airflow 側のジャーナル（state store か cache path） LLM プロバイダー側のキャッシュ XCom（キーは message_history） 再利用するときの動き 記録済みのステップはLLMもツールも呼ばずに結果を再生する LLMは毎回呼ぶ。入力トークンが割引されるだけ 過去の会話を前提にして、新しくLLMを呼ぶ コストへの効果 再生したステップは0（usage にも数えない） 入力トークン料金の一部が割引 減らない（履歴の分だけ入力が増える） 既定値 False True None（1回で終わる会話） ※表はAIが生成 実装例.サンプルデータの自動作成 今回以下のようなサンプルデータの自動生成と確認をしてくれる実装を試してみる。 seed_orders：ローカルDBに接続するためにSQLToolset、サンプルデータ作成時の注意点を考慮させるためにAgentSkillsToolsetを持たせます。 report：ローカルDBに接続するためにSQLToolset、slackに接続するためにHookToolset(SlackWebhookHook..)を持たせます。 また事前にpostgressへのconnectorをpostgres_default、slackへのconnectorをslack_webhook_defaultとして作成しておきます。 skills/seed-orders/SKILL.md --- name: seed-orders description: ケーキ屋の orders テーブルに初期データ(テスト用の注文)を投入するときの条件と手順。初期データの投入を頼まれたときに使う。 --- 初期データ投入の注意点 ・・・ dag.py from pathlib import Path from airflow.providers.common.ai.toolsets import HookToolset from airflow.providers.common.ai.toolsets.skills import AgentSkillsToolset from airflow.providers.common.ai.toolsets.sql import SQLToolset from airflow.providers.common.compat.sdk import dag, task from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook @dag(tags=["example"]) def example_agent_decorator(): create_orders = SQLExecuteQueryOperator( task_id="create_orders", conn_id="postgres_default", sql=[ "DROP TABLE IF EXISTS orders", """ CREATE TABLE orders ( id serial PRIMARY KEY, customer text, product text, category text, amount numeric(10, 2), created_at date ) """, ], ) @task.agent( llm_conn_id="my_claude_conn", system_prompt="あなたはデータエンジニアです。ツールを使ってデータベースを操作してください。", toolsets=[ SQLToolset( db_conn_id="postgres_default", allowed_tables=["orders"], allow_writes=True, ), AgentSkillsToolset( sources=[str(Path(file).parent / "skills")], exclude_tools={"run_skill_script"}, ), ], ) def seed_orders(): return "seed-orders スキルの手順に従って、orders テーブルに初期データを入れてください。" @task.agent( llm_conn_id="my_claude_conn", system_prompt="あなたはデータアナリストです。ツールを使って質問に答え、回答は日本語で書いてください。", toolsets=[ SQLToolset( db_conn_id="postgres_default", allowed_tables=["orders"], ), HookToolset( SlackWebhookHook(slack_webhook_conn_id="slack_webhook_default"), allowed_methods=["send_text"], tool_prefix="slack", ), ], ) def report(question: str): return ( "orders テーブルは、ケーキ屋の検証用に自動生成した注文データです。\n" "customer は顧客、product と category は買ったもの(ケーキ・焼き菓子・ドリンク)です。\n" f"このデータを使って、次の質問に答えてください: {question}\n" "Slack に1回だけ送ってください。送ったあと、送った内容をそのまま答えてください。" ) create_orders >> seed_orders() >> report("カテゴリ別の売上と、売れ筋の商品、売上が特に多かった月を教えてください。") example_agent_decorator() 実行してみると、途中でLLMがSQLを組み立てて考えてくれているログが見れる。 ・・・ 2026-10-25T05:58:42.234507Z [info ] ::group::Tool call: query [airflow.task.operators.airflow.providers.common.ai.decorators.agent._AgentDecoratedOperator] loc=logging.py:64 2026-10-25T05:58:42.243900Z [info ] Running statement: SELECT count(*) AS n, min(created_at) AS min_date, max(created_at) AS max_date, count(DISTINCT customer) AS customers, min(amount) AS min_amt, max(amount) AS max_amt, sum(CASE WHEN created_at BETWEEN DATE '2025-12-20' AND DATE '2025-12-25' THEN 1 ELSE 0 END) AS dec_rush_rows, sum(CASE WHEN amount % 50 <> 0 THEN 1 ELSE 0 END) AS non50_amounts FROM orders, parameters: None [airflow.task.hooks.airflow.providers.postgres.hooks.postgres.PostgresHook] loc=sql.py:896 ・・・ また完了するとDatabaseへデータの挿入が行われ、SlackへSummary Reportされていることが確認できる。 id | customer | product | category | amount | created_at -----+-------------+--------------------+----------+---------+------------ 1 | customer_3 | チョコレートケーキ | ケーキ | 650.00 | 2026-06-23 2 | customer_3 | マドレーヌ | 焼き菓子 | 150.00 | 2025-11-27 3 | customer_3 | 紅茶 | ドリンク | 550.00 | 2026-09-09 4 | customer_1 | コーヒー | ドリンク | 400.00 | 2026-07-23 5 | customer_7 | チーズケーキ | ケーキ | 500.00 | 2026-07-02 6 | customer_1 | ショートケーキ | ケーキ | 650.00 | 2025-12-21 ・・・ LLM Retry Policy と ClassifierRetryPolicy ! Airflow>=3.3.0が必要 Airflow3.3.0からRetry policiesというのができた。これを用いれば、LLMにタスクの失敗時のリトライ判断を委ねることができる。 種類としては、LLMRetryPolicyとClassifierRetryPolicyがサポートされ、いずれも以下のようにタスクのretry_policyに指定して使用する。 snowflake_policy = ClassifierRetryPolicy(・・・) ・・・ snowflake_policy = LLMRetryPolicy(・・・) ・・・ @task(retries=5, retry_policy=snowflake_policy) def query_snowflake(): ・・・ ではLLMRetryPolicyとClassifierRetryPolicyで何が異なるのかというと以下である。 LLMRetryPolicy 責務：モデルがRetryに関連するエラーの分類・動作をすべて決める。 ClassifierRetryPolicy 責務：モデルは分類だけ、動作はコードで定義した表で決める。 snowflake_policy = ClassifierRetryPolicy( ・・・ min_confidence=0.8, # categoriesの挙動を実行を決めるときの分類のmin確信度 categories={ Warehouseが停止したと分類したときの挙動 "warehouse_suspended": ErrorCategory( "The warehouse is suspended and will auto-resume.", delay=timedelta(seconds=30) ), tokenがexpiredしたと分類したときの挙動 "token_expired": ErrorCategory( "A JWT or session token expired; the token rotates on its own.", delay=timedelta(seconds=30) ) ・・・ またいずれもfallback_rulesでモデルが判断(分類)できなかったときのfallbackを設定できる。 実装例.モデルに再実行を委ねてみる rate limitエラーが出た場合と、bad dataエラーが出た場合に、モデルはどのように再実行を行うのかを試してみる。 from datetime import timedelta from airflow.providers.common.ai.policies.retry import LLMRetryPolicy from airflow.providers.common.compat.sdk import dag, task from airflow.sdk.definitions.retry_policy import RetryAction, RetryRule llm_policy = LLMRetryPolicy( llm_conn_id="my_claude_conn", timeout=30.0, fallback_rules=[ RetryRule(exception=ConnectionError, action=RetryAction.RETRY, retry_delay=timedelta(seconds=10)), RetryRule(exception=PermissionError, action=RetryAction.FAIL), ], ) @dag(catchup=False, tags=["sample", "retry_policy", "llm"]) def llm_retry_policy_sample(): @task(retries=3, retry_delay=timedelta(minutes=1), retry_policy=llm_policy) def rate_limited(): # rate limit エラー raise RuntimeError("429 Too Many Requests: Rate limit exceeded. Retry after 60 seconds.") @task(retries=3, retry_delay=timedelta(minutes=1), retry_policy=llm_policy) def bad_data(): # bad data エラー raise ValueError("Column 'user_id' expected type INT but got STRING in row 42.") rate_limited() bad_data() llm_retry_policy_sample() 実行してみると、rate limitエラーの方だけRetryが4回された。 人間としても、bad dataエラーの場合は再実行しても意味がない・rate limitエラーの場合は再実行したら直るかもと思うので、挙動としては正しいであろう。 ちなみに、LLMRetryPolicyにinstructionsが指定されていない場合、DEFAULT_INSTRUCTIONSが参照されるようだった。 そのほかにも そのほかにも以下がサポートされている。 @task.llm_branchを用いてLLM駆動型タスク分岐。 後続にA or Bのタスクがある場合、どちらを実行するかをLLMが決めてくれる。 @task.llm_sqlを用いて自然言語->sqlをbuildするタスクの作成。 sqlglotのAST解析を使って出力を検証している。 @task.llm_schema_compareを用いてDatabaseを横断したスキーマドリフト検出。 詳細は次のページを参照すれば確認できる。 airflow.apache.org どういった活用の仕方がされているか。 Airflow公式ブログ(link]では次のような活用の仕方が紹介がされている。 問い合わせチケットの振り分け（@task.llm_branch） レポートの要約と人による承認（@task.llm_file_analysis + enable_hitl_review=True） エージェントによるDB分析（@task.agent + SQLToolset）

Asked for a refund, got billing info? Try the OpenAI Agents SDK OpenAI Agents SDK is an open-source (MIT) Python framework for building AI agents and multi-agent workflows from a few parts: Agent, Runner, handoffs, guardrails, sessions and tracing. A triage agent can pass a support chat to a billing or refund specialist instead of one bot doing everything. It uses OpenAI models by default and other providers through OpenAI-compatible APIs or beta adapters (LiteLLM, Any-LLM). In this video: - The problem: a support bot answers a refund question with billing info - How it works: an Agent (model, instructions, tools) and the Runner loop - 3 real uses from the repo and docs: handoffs, an input guardrail, sessions - The catch: tracing defaults, other providers, model costs, version 0.x - Who it's for, an alternative, and how to try it How it works: an Agent is a model with instructions and tools. Runner.run() calls the model, runs tool calls, follows handoffs and loops until there is a final output. To the model, a handoff is a tool (transfer_to_refund_agent), and the agent that takes over sees the conversation. Three real uses (from the repo and docs): 1. Handoffs: examples/customer_service/main.py, an airline example where a Triage Agent hands questions to an FAQ Agent or a Seat Booking Agent, and they can hand back. 2. Guardrails: examples/agent_patterns/input_guardrails.py. A guardrail agent checks for math homework; the tripwire raises InputGuardrailTripwireTriggered and the run stops. By default an input guardrail runs alongside the agent, which may already have used tokens or tools; run_in_parallel=False makes it finish first. 3. Sessions: SQLiteSession("conversation_123") loads the history before each run and saves new items after, so "What state is it in?" follows on. Redis, SQLAlchemy, MongoDB and other backends exist. The catch: Tracing is on by default and exports to OpenAI's backend, including model and tool inputs and outputs (trace_include_sensitive_data defaults to True). Turn it off with OPENAI_AGENTS_DISABLE_TRACING=1, set_tracing_disabled(True) or per run, or replace the exporter with set_trace_processors(). Tracing is unavailable under an OpenAI Zero Data Retention policy. Other providers: OpenAI-compatible endpoints work through the built-in client settings; LiteLLM and Any-LLM are best-effort beta adapters. Without an OpenAI key, the docs say to disable tracing or use another trace processor. Hosted tools (web search, file search, code interpreter and more) run on OpenAI's servers with OpenAI Responses models; many other providers lack them. Every model call is billed by your provider. The SDK itself is MIT licensed. Still 0.x: v0.24.0 shipped 2026-10-15 (124 PyPI releases since 2025-03-04). The docs say minor versions can bring breaking changes, so pin your version. Who it's for: Python developers (3.10 or newer) who want a few building blocks in plain Python. For one short model call, the docs suggest the Responses API on its own. Want your whole agent type-checked? Look at Pydantic AI. The SDK also has a JavaScript/TypeScript version. Try it: pip install openai-agents (or uv add openai-agents) set OPENAI_API_KEY to your own key follow the quickstart: a Triage Agent with two specialist agents Links: Repo: openai/openai-agents-python Docs: openai.github.io Handoffs: openai.github.io Guardrails: openai.github.io Sessions: openai.github.io Tracing: openai.github.io Models and providers: openai.github.io Alternative: Pydantic AI, pydantic/pydantic-ai Facts checked 2026-10-21 (openai-agents 0.24.0, 30,202 GitHub stars). An open-source (MIT) option to consider, not an endorsement. Not affiliated with OpenAI, or with the Pydantic AI, LiteLLM or Any-LLM projects. Narration: AI voice. #OpenAIAgentsSDK #AIAgents #Python Chapters 0:00 Intro 0:13 The fix 0:30 How it works 0:47 Three real uses 1:09 The catch 1:33 Who it's for 1:46 Try it

【一次資料】 ■ 米国の実際の裁判例（2023年・Mata v. Avianca） law.justia.com ■ Pydantic AI 機能追加・修正・マージPR pydantic-ai#7434 ■ 私の原指摘：出典URLの取り違え pydantic-ai#7434 ■ 私の別の指摘：出典の数と独立した根拠の違い pydantic-ai#7434 ■ 開発者による修正と回帰テスト github.com/pydantic/pydantic-ai/commit/6594475317…

Prompt Caching in Pydantic AI: Up to 89% Lower Input Costs This article is for AI developers who want to build an AI agent with Pydantic AI that uses changing instructions and still saves costs. The problem is that Pydantic AI used to put static and dynamic instructions in the same block, so one small change broke the whole cache. Below, I show how it works today. The problem Static instructions stay the same for a long time, for example what the agent should do and which rules it has to follow. Dynamic instructions change depending on the situation, for example the current date. Before my pull request (PR), every time the date changed, the agent had to pay the full price for all static and dynamic instructions again, instead of reading them from the cache for 10% of the price, even though only one dynamic instruction had changed. I found this problem by looking for an issue the maintainers wanted, but nobody had worked on yet. Issue #4543 had no replies, and a related issue about better Anthropic caching (#3453) had been open since November 2025. So I opened PR #4865. How prompt caching works on Anthropic Anthropic caches the beginning of a request up to a breakpoint that you mark with cache_control. The next request can only read from the cache if everything up to this breakpoint is exactly the same. The order is always tools first, then the system prompt, then the messages. A change at one of these levels makes the cache for this level and all levels after it invalid. You can set at most 4 breakpoints per request and choose between a lifetime of 5 minutes or 1 hour. Writing to the cache costs 25% more than the normal input price (5 minutes) or twice as much (1 hour), but reading from it only costs 10%. Static and dynamic instructions in Pydantic AI The code below is a simple example of static and dynamic instructions: from datetime import date from pydantic_ai import Agent, RunContext POLICY = "..." agent = Agent( 'anthropic:claude-haiku-5-5', deps_type=str, instructions=POLICY, # static: a long return policy that never changes ) @agent.instructions def customer_context(ctx: RunContext[str]) -> str: # dynamic: changes per request return f"Customer name: {ctx.deps}. Today's date: {date.today()}." As the comments show, POLICY is the static instruction and the function customer_context is the dynamic instruction. Pydantic AI treats everything you pass to instructions= as static and every function with @agent.instructions as dynamic. date.today() depends on the current date and the customer name comes from the request, so both change. POLICY stays the same unless you change the policy yourself. My first approach and what the review changed My first thought was to create a new parameter add_cache_breakpoint for @agent.instructions and a new setting anthropic_static_cache_instructions, because not everyone needs this kind of caching with Anthropic. But this setting would be in the core of the framework, so every user, including users who work with other AI providers like OpenAI or Google, would see it, although it does nothing for them. So the maintainer, Douwe Maan, came up with the idea to make it automatic instead of adding a new setting and parameter, and reworked the PR. Now static instructions are always sorted before dynamic ones, and with the existing setting anthropic_cache_instructions=True, the cache breakpoint is placed automatically after the last static instruction. I loved the idea. You don't need a new setting, you don't have to read about it or even know about it, because you already have it. Working together with the maintainer of Pydantic AI helped me a lot to learn about discussions and architectural design. It also taught me to keep looking for the best solution until I'm really sure I have it. How to use it today To cache the static instructions, you only have to add one setting to the agent from above: from pydantic_ai.models.anthropic import AnthropicModelSettings agent = Agent( 'anthropic:claude-haiku-5-5', deps_type=str, instructions=POLICY, model_settings=AnthropicModelSettings(anthropic_cache_instructions=True), ) Instead of AnthropicModelSettings(anthropic_cache_instructions=True), you can also use model_settings={'cache': True}. This is the provider-independent setting for caching. On Anthropic, it also caches the tool definitions and the conversation. To check if it works, I ran the agent three times with a different customer each time: for customer in ['Alice', 'Bob', 'Carol']: result = agent.run_sync('Can I return unused headphones from category 7?', deps=customer) usage = result.response.usage print(customer, usage.input_tokens, usage.cache_write_tokens, usage.cache_read_tokens) These are the results with Claude Haiku 5.5 and Pydantic AI 2.60.0: | Run | Input tokens | Cache write | Cache read | |-------|-------------:|------------:|-----------:| | Alice | 3,247 | 3,205 | 0 | | Bob | 3,247 | 0 | 3,205 | | Carol | 3,247 | 0 | 3,205 | In the first run with Alice, the static policy is written to the cache. In the runs with Bob and Carol, 3,205 of 3,247 input tokens come from the cache, although the customer name changed. Only the remaining 42 tokens, the dynamic instructions and the question, are charged at the normal price. For the costs, this means that the first run is about 25% more expensive, because writing to the cache costs extra. But every run after that costs about 89% less for the input. Over these three runs, that already saves about 51% of the input costs, and the more requests you have, the more you save. Pitfalls There are a few things you should pay attention to: - Don't use dynamic values in static instructions. Agent(instructions=f'... {date.today()}') looks static, but it's not, because the date changes every day. So the cache is gone every day. Put values like this in an @agent.instructions function instead. - Dynamic instructions that change during a conversation. If a dynamic part changes in the middle of a conversation, the static part stays in the cache, but the cached conversation after it is gone. To handle this, use @agent.instructions(name='state', on_change='append'). Then each change is appended to the message history instead of rewriting the instructions. The history gets a bit longer with every change, but the cache stays. - Instructions that are too short. Caching only works above a minimum length, for example 512 tokens for Claude Haiku 5.5 and up to 4,096 tokens for older models. If your instructions are shorter, nothing is cached and you don't get an error. To catch that, check cache_write_tokens and cache_read_tokens. If both are 0, nothing got cached. Conclusion Prompt caching only works if the beginning of your prompt stays exactly the same. So keep your static instructions static and put everything that changes in an @agent.instructions function. Pydantic AI then sorts the static instructions before the dynamic ones and places the cache breakpoint between them automatically. You only need one setting for it. In my test, every request after the first one cost about 89% less for the input. My biggest lesson from this PR: always ask if there is an even simpler way for the developer who uses your code, without making it unnecessarily complex. You can find all caching options in the Pydantic AI docs. AI helped me with the English, the wording and some parts of the text. I reviewed and edited everything myself.

LLMの評価を楽にしたいので、Pydantic Evalsを使ってみる はじめに ここ1年ほど、LLMの調整をする中で、特に大変だったのが品質の確認です。調整のたびに結果を評価し直す必要があり、その繰り返しがしんどくなることも多くありました。評価用ライブラリがいくつも登場しているのも、同じような苦労をしている開発者が多いからではないかと思います。 また、JevのようなDecisionモデルも出てきているので、モデルの判断を数値で確かめることは、これからさらに大事になると自分は思っています。評価ライブラリを使って、変更前後の結果を比較できるようにしておきたいです。 今回は、Pydantic Evalsで問い合わせ分類を評価してみます。Pydantic AIと組み合わせる例や、ChatGPTの利用枠でLLMJudgeを使う例も載せています。 Pydantic Evalsとは Pydantic Evalsは、LLMアプリやAIエージェントの出力を評価するためのPython製フレームワークです。評価用の入力と採点ルールを用意して処理を動かすと、ケースごとの出力や合否、スコアなどをレポートで確認できます。 問い合わせ分類なら正解ラベルと合っているか、回答文なら質問に答えているかを評価できます。通常のPython関数も評価対象にできるため、LLMを呼ばずに評価の流れを試せます。 Pydanticで型や構造を確認できても、分類先や回答の中身まで正しいとは限りません。その部分を確かめるのがPydantic Evalsの役割です。 評価ライブラリを使う理由 LLMは同じ入力でも出力が変わることがあるので、一度うまく答えられただけでは安定して使えるかが分かりません。用途に合った評価基準を決めて、繰り返し確認する必要があります。 プロンプトやモデルの変更によって、以前はできていたことができなくなる場合もあります。このような変更に伴う品質の低下を、回帰（regression）と呼びます。1 エージェントの評価では、回答文だけでなく、必要なツールを使ったか、依頼された処理を完了したかも確認します。 評価データを管理し、実行や採点、結果の比較まで毎回自分で用意するのは大変です。DeepEvalやRagasなどもありますが、今回はPydantic Evalsに任せてみます。 自分がPydantic Evalsを選ぶ理由 自分がPydantic Evalsを選んでいるのは、Pydanticを中心としたエコシステムが整っているからです。PydanticはPythonのデータ検証で広く使われており、自分としてはデファクトスタンダードと捉えています。 その周辺には、AIアプリケーションを開発するPydantic AI、処理の分岐や状態遷移をグラフとして定義するPydantic Graph、ログやトレースを使って実行状況を調べるLogfireもあります。 これらを組み合わせて、実装から評価、運用時の調査まで進められるのがよいと思っています。Pydantic Evalsも、その中で積極的に使っていくつもりです。 基本的な構成 コードに出てくるものを先に整理しておきます。 要素 役割 問い合わせ分類の例 Case 1件の評価データ 入力文と期待する分類先 Dataset Caseと共通の採点ルールをまとめるもの 問い合わせの評価セット Task 評価対象の処理 入力文から分類先を返す関数 Evaluator 結果を採点するルール 期待する分類先との一致判定 EvaluationReport 評価実行の結果 各ケースの出力や合否 評価対象は普通のPython関数です。Datasetのevaluate_syncに関数を渡して実行します。 とりあえず使ってみる まずは、問い合わせを分類する簡単な関数で動かしてみます。LLMは呼ばず、返ってきた分類先が期待したものと合っているかを評価します。 環境 OS：Windows Python：3.12.13 pydantic-evals：2.58.0 uvで環境を作成します。 uv init --python 3.12 --vcs none pydantic-evals-intro cd pydantic-evals-intro uv add "pydantic-evals==2.58.0" モデルのAPIキーやLogfireのアカウントは不要です。 コード例 今回は、問い合わせをrefund、account、otherの3種類に分類します。返金とアカウントの両方に触れている場合は、返金を優先するルールにします。 プロジェクト直下にmain.pyを作成し、以下を記述します。 main.py from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import EqualsExpected dataset = Dataset[str, str]( name="support_classification", cases=[ Case( name="refund", inputs="返金してほしいです", expected_output="refund", ), Case( name="account", inputs="ログインできません", expected_output="account", ), Case( name="other", inputs="営業時間を教えてください", expected_output="other", ), Case( name="mixed_intent", inputs="ログインはできますが、返金してほしいです", expected_output="refund", ), ], evaluators=[EqualsExpected()], ) def classify_v1(text: str) -> str: if "ログイン" in text: return "account" if "返金" in text: return "refund" return "other" if name == "main": report = dataset.evaluate_sync(classify_v1) report.print( include_input=True, include_output=True, include_expected_output=True, ) Caseに入力と正解ラベルを入れ、EqualsExpectedで実際の出力と比較します。expected_outputを入れるだけでは採点されないので、evaluatorsへの指定も必要です。 実行結果 main.pyを実行します。 uv run python -X utf8 main.py Windowsでの文字化けを避けるため、-X utf8を付けています。 実行すると、ケースごとの出力や合否が表示されます。Assertionsが合否の欄です。結果を抜き出すと、以下のようになります。 Case 期待する出力 実際の出力 EqualsExpected refund refund refund 成功 account account account 成功 other other other 成功 mixed_intent refund account 失敗 4件中3件が成功し、この評価データでの成功率は75%でした。mixed_intentは返金の依頼ですが、「ログイン」という単語に先に反応してaccountになっています。 採点が不合格でもスクリプトは正常終了するので、レポートで結果を確認します。 修正して比較する 返金を先に判定するように修正します。評価データは変えずに、変更前と変更後を比べてみます。 main.py末尾のifブロックを、次のコードに置き換えます。CaseやDataset、classify_v1の定義はそのまま使います。 main.py def classify_v2(text: str) -> str: if "返金" in text: return "refund" if "ログイン" in text: return "account" return "other" if name == "main": baseline = dataset.evaluate_sync(classify_v1) candidate = dataset.evaluate_sync(classify_v2) candidate.print(baseline=baseline, include_output=True) main.pyをもう一度実行します。baseline引数に変更前のレポートを渡すと、変更後の結果と比較できます。 この修正ではmixed_intentも成功し、成功率は75%から100%になりました。他の3件の結果も変わっていません。 ただ、今の関数だと「返金は不要ですが、ログインできません」もrefundになります。100%になったのは用意した4件についてで、ほかの問い合わせまで正しく分類できるとは限りません。 Pydantic AIと組み合わせる 次はPydantic AIでLLMを呼び、同じ4件を分類する形にします。DatasetとEqualsExpectedは、そのまま使えます。 OpenAIのモデルを使うので、APIキーが必要です。テスト用モデルで評価までの流れは確認していますが、実APIでの結果は未確認です。 準備 同じプロジェクトに、プロバイダー用のパッケージを追加します。 uv add "pydantic-ai-slim[openai]==2.58.0" 今回はGPT-6 Lunaを指定します。PowerShellでAPIキーとモデル名を設定します。 $env:OPENAI_API_KEY = "<自分のAPIキー>" $env:EVAL_MODEL = "openai:gpt-6-luna" エージェントのコード main.pyと同じ場所にllm_eval.pyを作成します。 llm_eval.py import os from typing import Literal from pydantic_ai import Agent from main import dataset agent = Agent( os.environ["EVAL_MODEL"], output_type=Literal["refund", "account", "other"], instructions=( "問い合わせを分類してください。" "返金の依頼はrefund、ログインやアカウントの問題はaccount、" "それ以外はotherです。両方の依頼があれば返金を優先してください。" "単語の有無だけでなく、依頼の内容を読んで判断してください。" ), ) async def classify_with_llm(text: str) -> str: result = await agent.run(text) return result.output if name == "main": report = dataset.evaluate_sync(classify_with_llm) report.print(include_output=True, include_expected_output=True) Agentのoutput_typeにLiteralを指定して、出力候補を3種類に絞っています。分類先が入ったresult.outputを返し、EqualsExpectedで正しい分類先を選べたかを見ています。 実行 llm_eval.pyを実行します。 uv run python -X utf8 llm_eval.py evaluate_syncには、この例のようなasync関数も渡せます。 LLMJudgeで回答文を評価する 分類なら正解ラベルと一致しているかを見ればよいですが、回答文では表現が違っても内容は正しいことがあります。こうした回答を評価するのに、LLMJudgeを使ってみます。 組み込みのEvaluatorには、以下のようなものがあります。 Evaluator 確認すること 使う場面 EqualsExpected ケースの期待する出力との一致 分類ラベルなど Contains 指定した文字列などを含むか 必須の語句があるかの確認 IsInstance 出力の型名 文字列や辞書などの型の確認 LLMJudge 自然言語で指定した評価基準を満たすか 回答の関連性や説明の十分さなど 採点基準 営業時間の回答を例に、rubricに以下の基準を渡します。 営業時間が期待する回答と一致すること 期待する回答にない定休日や受付条件を付け加えないこと 「平日9時から17時です」と「平日の午前9時から午後5時まで営業しています」のような言い換えは許容する想定です。 この基準で採点するため、実際の出力に加え、include_input=Trueで質問を、include_expected_output=Trueで期待する回答も渡します。 LLMJudgeはデフォルトで合否を返します。report.printにinclude_reasons=Trueを指定すると、判定理由も見られます。 採点する側もLLMなので、判定が揺れることがあります。人が付けた判定と照らし合わせて、意図した基準で採点されているかを確認しながら調整します。 ChatGPTの利用枠を使う 採点に使うモデルはmodel引数で指定します。今回はSign in with ChatGPTを使い、対象となるChatGPT Plus／Proアカウントの利用枠で採点する構成にします。 Pydantic AIのChatGPTプロバイダーでブラウザ認証し、そのモデルをLLMJudgeに渡します。今回の2.58.0でも利用できます。 この仕組みはプレビュー提供で、対象はOSSやローカルで動かすアプリです。有料アプリやリモートで提供するアプリで使う場合は、OpenAIへの申請・承認が必要です。サブスクリプションの利用上限やアカウントの利用資格も適用されるため、利用条件とプレビューの制限を確認してください。 パッケージは先ほど追加したものを使います。この節の採点はChatGPTの利用枠を使い、OPENAI_API_KEYとEVAL_MODELは使いません。先ほどの分類例はAPI側の課金対象です。 採点用のモデル名をJUDGE_MODELに設定します。以下はPowerShellの例です。 $env:JUDGE_MODEL="gpt-6.1-sol" モデル名はPydantic AIの公式例に合わせています。アカウントで利用できるモデルに置き換えてください。 judge_eval.pyを作成します。回答は固定し、採点するときだけモデルを呼びます。 judge_eval.py import asyncio import json import os import uuid import webbrowser from pathlib import Path from pydantic_ai.models.openai import OpenAIResponsesModel from pydantic_ai.providers.openai_chatgpt import ( OpenAIChatGPTOAuthFlow, OpenAIChatGPTProvider, ) from pydantic_evals import Case, Dataset from pydantic_evals.evaluators import LLMJudge def fixed_answer(question: str) -> str: return "平日の午前9時から午後5時まで営業しています。" async def main() -> None: model_name = os.environ["JUDGE_MODEL"] registration_file = Path(".chatgpt-registration.json") if not registration_file.exists(): registration_file.write_text( json.dumps({"host_id": uuid.uuid4().urn}), encoding="utf-8" ) registration = json.loads(registration_file.read_text(encoding="utf-8")) flow = OpenAIChatGPTOAuthFlow( app_name="Pydantic Evals Intro", host_id=registration["host_id"], client_id=registration.get("client_id"), ) webbrowser.open(flow.authorization_url(extra_params={"prompt": "consent"})) credentials = await flow.exchange_code_from_callback() registration["client_id"] = credentials.client_id registration_file.write_text(json.dumps(registration), encoding="utf-8") if not credentials.plan_usage_granted: raise RuntimeError("ChatGPTの利用枠を使う許可が必要です。") provider = OpenAIChatGPTProvider(credentials=credentials) model = OpenAIResponsesModel(model_name, provider=provider) judge = LLMJudge( model=model, rubric=( "回答が質問に直接答えており、営業時間が期待する回答と一致すること。" "期待する回答にない定休日や受付条件を付け加えていないこと。" ), include_input=True, include_expected_output=True, ) hours_dataset = Dataset[str, str]( name="business_hours", cases=[ Case( name="paraphrased_answer", inputs="営業時間は？", expected_output="平日9時から17時です", ), ], evaluators=[judge], ) report = await hours_dataset.evaluate(fixed_answer) report.print( include_input=True, include_output=True, include_expected_output=True, include_reasons=True, ) if name == "main": asyncio.run(main()) judge_eval.pyを実行します。 uv run python -X utf8 judge_eval.py ブラウザが開いたらChatGPTにログインし、利用枠の使用を許可します。ブラウザとスクリプトは同じPCで動かしてください。認証には127.0.0.1の1455番ポートを使います。 .chatgpt-registration.jsonには登録情報を保存します。トークンは保存しないので、実行のたびにブラウザ認証が必要です。認証情報も保存する場合は、公式ドキュメントのOpenAIChatGPTCredentialSourceを使い、トークンをGitやログに残さず、本人だけが読み書きできる場所で管理します。 認証処理とHTTP応答をテスト用のものに差し替えて、評価までの流れは確認しています。実アカウントでの認証や利用枠の消費、採点結果は未確認です。 継続して使う場合 判定に迷う入力や過去に失敗した入力も評価データに追加していきます。全体の成功率だけでなく、失敗したケースも見ておきます。 比較には同じ評価データと採点ルールを使い、モデル名やパッケージのバージョンも残しておきます。 出力の揺れも見たい場合は、repeatで各ケースを複数回実行できます。llm_eval.pyでレポートを作る行を以下のように変えると、4ケースを3回ずつ、計12回実行します。 report = dataset.evaluate_sync(classify_with_llm, repeat=3) モデルのリトライがあれば、APIへのリクエスト数はさらに増えます。費用や利用枠を見ながら回数を決め、必要ならmax_concurrencyで同時実行数を抑えます。 おわりに 4件の問い合わせでも、評価データを用意しておくと、修正でどの結果が変わったかを確認できました。LLMを使う場合も同じDatasetで比較でき、回答文にはLLMJudgeを使えます。 普段の確認に使っている入力をCaseとして残し、失敗したケースも追加しながら、変更前後の比較に使っていこうと思います。 参考 Pydantic Evals Overview Quick Start Core Concepts Built-in Evaluators LLM Judge pydantic-evals 2.58.0 / PyPI 脚注 ここでのregressionは、変更によって以前の機能・品質が損なわれることを指すソフトウェア開発の用語で、統計学の回帰分析（regression analysis）とは意味が異なります。LangSmith公式のRegression testsの説明も参照してください。 ↩︎

Pydantic AI advisory: with tracing on, provider-native MCP tools could put server URLs, auth tokens and headers into OpenTelemetry traces, even with include_content=False. Upgrade to 2.58.0 / 1.107.8, then lock down old traces and rotate those tokens. github.com/pydantic/pydantic-ai/security/advisori…