「MCP is all you need」Pydantic作者が語るMCPの正しい使い方とsampling完全解説

# pypi_mcp_server.py
from mcp.server.fastmcp import FastMCP
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from google.cloud import bigquery

mcp = FastMCP("pypi-stats-server")

@mcp.tool()
async def pypi_downloads(
    ctx: MCPContext,
    question: str,
) -> str:
    """PyPIパッケージのダウンロード統計をBigQueryで照会する。
    
    ユーザーの自然言語の質問をBigQuery SQLに変換し、
    PyPI公開データセットを照会してダウンロード数を返す。
    """
    # MCPコンテキスト経由でログ送信（ツール実行中にクライアントへ返る）
    await ctx.log("info", f"BigQuery照会を開始: {question}")
    
    # 内部でPydanticAIエージェントを動かす（Samplingを使用）
    result = await _run_query_agent(ctx, question)
    return result

async def _run_query_agent(ctx: MCPContext, question: str) -> str:
    """内部エージェント：クライアントのLLMをSampling経由で使用"""
    agent = Agent(
        model=ctx.sampling,  # メインエージェントのLLMを借用
        system_prompt=BIGQUERY_SYSTEM_PROMPT,
        result_type=QueryResult,
        result_validator=validate_and_run_query,
    )
    result = await agent.run(question)
    return format_as_xml(result.data)

if __name__ == "__main__":
    mcp.run()  # デフォルトはstdio transport

from pydantic_ai import ModelRetry

async def validate_and_run_query(
    ctx: RunContext[MCPContext],
    result: str
) -> str:
    """生成されたSQLを検証・実行。失敗時はModelRetryでLLMに再生成を依頼"""
    # LLMがmarkdownコードブロックで囲んだ場合は除去
    sql = strip_markdown_fences(result)
    
    # テーブル名の検証
    if WRONG_TABLE in sql:
        raise ModelRetry(
            f"テーブル名が間違っています。正しくは: {CORRECT_TABLE}"
        )
    
    # BigQueryでSQLを実際に実行
    client = bigquery.Client()
    try:
        rows = list(client.query(sql).result())
        return format_rows_as_xml(rows)
    except Exception as e:
        # 実行失敗時は自動リトライ
        raise ModelRetry(f"SQLエラー: {e}\n再度試してください。")

# main_agent.py
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio
from datetime import date

# MCPサーバーをサブプロセスとして起動・登録
research_agent = Agent(
    model="openai:gpt-4o",
    mcp_servers=[
        MCPServerStdio("python", ["pypi_mcp_server.py"])
    ],
    system_prompt=f"""
    あなたはPyPIパッケージの調査エージェントです。
    今日の日付: {date.today()}
    """,
)

# 実行
result = await research_agent.run(
    "Pydanticは今年何回ダウンロードされましたか？"
)
print(result.data)
# → "Pydantic は 2026年に約 1.6 billion 回ダウンロードされています。"

@mcp.tool()
async def long_running_research(ctx: MCPContext, query: str) -> str:
    await ctx.log("info", "データソースを検索中...")
    results_a = await search_source_a(query)
    
    await ctx.log("info", f"ソースAから{len(results_a)}件取得。分析中...")
    analysis = await analyze(results_a)
    
    # progressノーティフィケーションも使える
    await ctx.report_progress(current=50, total=100)
    
    await ctx.log("info", "最終レポートを生成中...")
    return generate_report(analysis)