Deferred Tools

from pydantic_ai import (
    Agent,
    DeferredToolRequests,
    DeferredToolResults,
    ToolDenied,
)
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    output_type=[str, DeferredToolRequests],
    instructions='ファイル操作の依頼を受けたら delete_file を呼んでください。',
)
 
@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    """指定パスのファイルを削除する (要承認)。"""
    # 承認後に呼ばれるとここに到達する
    return f'{path} を削除しました'
 
# 1 回目の run: 承認が必要なら DeferredToolRequests が返る
result = agent.run_sync('/tmp/old.log を削除してください')
 
if isinstance(result.output, DeferredToolRequests):
    # 人間にお伺いする処理 (UI / Slack / メール 等)
    requests = result.output
    results = DeferredToolResults()
    for call in requests.approvals:
        # ここでは UI 越しに承認を取った想定で True
        results.approvals[call.tool_call_id] = True
        # 拒否する場合は: ToolDenied('権限が不足しています')
 
    # 2 回目の run: message_history + deferred_tool_results で再開
    final = agent.run_sync(
        message_history=result.all_messages(),
        deferred_tool_results=results,
    )
    print(final.output)
else:
    print(result.output)

from pydantic_ai import (
    Agent, ApprovalRequired, DeferredToolRequests, DeferredToolResults,
    RunContext,
)
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    output_type=[str, DeferredToolRequests],
)
 
@agent.tool
def update_file(ctx: RunContext, path: str, content: str) -> str:
    """ファイルを書き換える。重要パスは承認必須。"""
    if path in {'.env', 'production.yml'} and not ctx.tool_call_approved:
        raise ApprovalRequired()
    return f'{path} を更新しました'

from pydantic_ai import (
    Agent, CallDeferred, DeferredToolRequests, DeferredToolResults,
    RunContext,
)
from pydantic_ai.models.google import GoogleModel
 
agent = Agent(
    GoogleModel('gemini-3-flash-preview'),
    output_type=[str, DeferredToolRequests],
)
 
@agent.tool
async def render_video(ctx: RunContext, prompt: str) -> str:
    """動画レンダリング (数分〜)。外部キューに enqueue して結果を後で返す。"""
    job_id = enqueue_render_job(prompt)  # 仮想関数: 別システムにジョブ投入
    raise CallDeferred(metadata={'job_id': job_id, 'prompt': prompt})
 
result = await agent.run('海辺の夕日の動画を作って')
 
if isinstance(result.output, DeferredToolRequests):
    requests = result.output
    results = DeferredToolResults()
    for call in requests.calls:
        meta = requests.metadata[call.tool_call_id]
        # 外部キューから結果を待つ (ここでは擬似的に取得)
        url = wait_for_render_job(meta['job_id'])
        results.calls[call.tool_call_id] = url
 
    final = await agent.run(
        message_history=result.all_messages(),
        deferred_tool_results=results,
    )
    print(final.output)

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import HandleDeferredToolCalls
 
async def handle_deferred(ctx, requests):
    return requests.build_results(approve_all=True, calls={
        c.tool_call_id: 'mock_result' for c in requests.calls
    })
 
agent = Agent(
    'google-gla:gemini-3-flash-preview',
    capabilities=[HandleDeferredToolCalls(handler=handle_deferred)],
)