AI エージェント開発実践 #6 MCP サーバーを自作する

pip install "mcp[cli]"

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("order-tools")

@mcp.tool()
def search_orders(query: str) -> str:
    """注文番号、顧客名、日付で注文を検索する。
    ユーザーが特定の注文の状態や内訳を尋ねたら、まずこのツールで注文を探す。"""
    results = do_search(query)          # 実際の実装
    return format_results(results)

@mcp.tool()
def get_order_detail(order_id: str) -> str:
    """注文番号で注文の詳細を照会する。例: ORD-2026-0001"""
    return fetch_detail(order_id)

if __name__ == "__main__":
    mcp.run()   # デフォルトは stdio 方式で実行される

import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

client = anthropic.AsyncAnthropic()

async def run_agent(goal: str, max_steps: int = 20) -> str:
    server = StdioServerParameters(command="python", args=["order_server.py"])
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as mcp_session:
            await mcp_session.initialize()

            # 1. サーバーのツール一覧を Claude の tools 形式に変換する
            listed = await mcp_session.list_tools()
            tools = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in listed.tools
            ]

            # 2. ループは第1回と同じ。ツールの実行だけ MCP 呼び出しに変わる。
            messages = [{"role": "user", "content": goal}]
            for _ in range(max_steps):
                response = await client.messages.create(
                    model="claude-opus-4-8",
                    max_tokens=16000,
                    tools=tools,
                    messages=messages,
                )
                if response.stop_reason == "end_turn":
                    return next(b.text for b in response.content if b.type == "text")

                messages.append({"role": "assistant", "content": response.content})
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = await mcp_session.call_tool(block.name, block.input)
                        text = "\n".join(c.text for c in result.content if c.type == "text")
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": text,
                            "is_error": bool(result.isError),
                        })
                messages.append({"role": "user", "content": tool_results})

    raise RuntimeError("ステップ上限内に作業を終えられませんでした。")

print(asyncio.run(run_agent("ORD-2026-0001 の注文状態を教えてください。")))