Advanced Tools

from pydantic import BaseModel
from pydantic_ai import Agent
 
class Weather(BaseModel):
    city: str
    condition: str
    temp_c: float
 
WEATHER_DB = {
    'Tokyo': Weather(city='Tokyo', condition='sunny', temp_c=22.0),
    'Sapporo': Weather(city='Sapporo', condition='snowy', temp_c=-2.0),
}
 
agent = Agent('google-gla:gemini-3-flash-preview', instructions='短く日本語で。')
 
@agent.tool_plain
def get_weather(city: str) -> Weather:
    """指定都市の天気を構造化して返します。"""
    if city not in WEATHER_DB:
        return Weather(city=city, condition='unknown', temp_c=0.0)
    return WEATHER_DB[city]
 
print(agent.run_sync('東京と札幌の天気を比較して').output)

import asyncio
from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext
 
@dataclass
class HttpDeps:
    http: httpx.AsyncClient
 
agent_p = Agent('google-gla:gemini-3-flash-preview', deps_type=HttpDeps)
 
@agent_p.tool
async def fetch_summary(ctx: RunContext[HttpDeps], url: str) -> str:
    """URL の冒頭 200 字を返します。"""
    r = await ctx.deps.http.get(url, timeout=5.0)
    return r.text[:200]
 
# LLM は 3 件のサマリを並列に取りに行く
async def main():
    async with httpx.AsyncClient() as http:
        result = await agent_p.run(
            '次の 3 サイトを比較して: https://a.example.com, https://b.example.com, https://c.example.com',
            deps=HttpDeps(http=http),
        )
        print(result.output)

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import ToolDefinition
 
@dataclass
class Auth:
    is_admin: bool
 
agent_a = Agent('google-gla:gemini-3-flash-preview', deps_type=Auth)
 
# 通常 tool
@agent_a.tool_plain
def list_users() -> list[str]:
    """ユーザー一覧を返します。"""
    return ['tanaka', 'sato']
 
# 管理者だけが使える tool
async def only_for_admin(ctx: RunContext[Auth], tool_def: ToolDefinition) -> ToolDefinition | None:
    return tool_def if ctx.deps.is_admin else None
 
@agent_a.tool_plain(prepare=only_for_admin)
def delete_user(name: str) -> str:
    """指定ユーザーを削除します (管理者専用)。"""
    return f'削除: {name}'
 
# 一般ユーザーで実行 — delete_user は schema に出ない
print(agent_a.run_sync('tanaka を削除して', deps=Auth(is_admin=False)).output)
 
# 管理者で実行 — delete_user が呼べる
print(agent_a.run_sync('tanaka を削除して', deps=Auth(is_admin=True)).output)

async def with_user_hint(ctx: RunContext[Auth], tool_def: ToolDefinition) -> ToolDefinition:
    role = '管理者' if ctx.deps.is_admin else '一般ユーザー'
    tool_def.description = f'{tool_def.description} (現在のロール: {role})'
    return tool_def
 
@agent_a.tool_plain(prepare=with_user_hint)
def show_status() -> str:
    """現在のシステム状態を返します。"""
    return 'OK'

result = agent.run_sync('東京の天気は?')
for msg in result.all_messages():
    for part in msg.parts:
        if hasattr(part, 'tool_name'):
            print(f'{msg.kind}: {part.tool_name}({getattr(part, "args", "")})')