MCP クライアント側の実装：Claude Desktop / API

MCP サーバーを作っても、それを呼び出すクライアント側がなければ動きません。本レッスンでは、Claude Desktop と API クライアントの両方の実装を学びます。

Claude Desktop での利用

設定ファイルの場所

設定ファイルの例

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxxx"
      }
    },
    "postgres": {
      "command": "python",
      "args": ["/path/to/postgres_mcp.py"],
      "env": {
        "DATABASE_URL": "postgresql://localhost/mydb"
      }
    },
    "company-tools": {
      "command": "python",
      "args": ["/path/to/company_mcp.py"],
      "env": {
        "API_KEY": "...",
        "USER_ID": "user_123"
      }
    }
  }
}

動作確認

1. 設定ファイルを保存
2. Claude Desktop を再起動
3. 右下の「⚙️」アイコンで MCP サーバーが登録されているか確認
4. 新規チャットで MCP ツールが認識されているか確認

Claude API（Python SDK）からの利用

# pip install anthropic mcp

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

async def chat_with_mcp(user_message: str):
    # MCP サーバー起動
    server_params = StdioServerParameters(
        command="python",
        args=["postgres_mcp.py"],
        env={"DATABASE_URL": "postgresql://localhost/mydb"},
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 利用可能なツール取得
            tools_response = await session.list_tools()
            mcp_tools = tools_response.tools

            # Claude APIフォーマットに変換
            claude_tools = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in mcp_tools
            ]

            # Claude を呼び出し
            client = anthropic.Anthropic()
            messages = [{"role": "user", "content": user_message}]

            while True:
                response = client.messages.create(
                    model="claude-sonnet-4-7",
                    max_tokens=4096,
                    tools=claude_tools,
                    messages=messages,
                )

                if response.stop_reason == "end_turn":
                    print(response.content[-1].text)
                    break

                if response.stop_reason == "tool_use":
                    # ツール呼び出し
                    tool_uses = [c for c in response.content if c.type == "tool_use"]
                    messages.append({"role": "assistant", "content": response.content})

                    tool_results = []
                    for tu in tool_uses:
                        # MCP サーバー経由で実行
                        result = await session.call_tool(tu.name, tu.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": tu.id,
                            "content": result.content[0].text,
                        })

                    messages.append({"role": "user", "content": tool_results})


asyncio.run(chat_with_mcp("ユーザーテーブルから直近1ヶ月の登録数を集計して"))

TypeScript SDK での利用

// npm install @anthropic-ai/sdk @modelcontextprotocol/sdk

import Anthropic from "@anthropic-ai/sdk";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function chatWithMcp(userMessage: string) {
    // MCP サーバー起動
    const transport = new StdioClientTransport({
        command: "node",
        args: ["./mcp-server.js"],
    });

    const mcpClient = new Client({
        name: "my-client",
        version: "1.0.0",
    }, {
        capabilities: {},
    });

    await mcpClient.connect(transport);

    const tools = await mcpClient.listTools();
    const claudeTools = tools.tools.map(t => ({
        name: t.name,
        description: t.description,
        input_schema: t.inputSchema,
    }));

    const anthropic = new Anthropic();
    const messages: any[] = [{ role: "user", content: userMessage }];

    while (true) {
        const response = await anthropic.messages.create({
            model: "claude-sonnet-4-7",
            max_tokens: 4096,
            tools: claudeTools,
            messages: messages,
        });

        if (response.stop_reason === "end_turn") {
            console.log(response.content);
            break;
        }

        if (response.stop_reason === "tool_use") {
            const toolUses = response.content.filter(c => c.type === "tool_use");
            messages.push({ role: "assistant", content: response.content });

            const toolResults = await Promise.all(
                toolUses.map(async (tu: any) => {
                    const result = await mcpClient.callTool({
                        name: tu.name,
                        arguments: tu.input,
                    });
                    return {
                        type: "tool_result",
                        tool_use_id: tu.id,
                        content: result.content[0].text,
                    };
                })
            );

            messages.push({ role: "user", content: toolResults });
        }
    }
}

複数MCPサーバーを同時に使う

# 複数MCP サーバーを統合

async def multi_mcp_agent(user_message):
    # 並列でMCPサーバーに接続
    servers = await asyncio.gather(
        connect_mcp("postgres_mcp.py"),
        connect_mcp("slack_mcp.py"),
        connect_mcp("notion_mcp.py"),
    )

    # 全ツールを集約
    all_tools = []
    server_for_tool = {}

    for server in servers:
        tools = await server.list_tools()
        for tool in tools.tools:
            all_tools.append({
                "name": f"{server.name}_{tool.name}",  # サーバー名付き
                "description": tool.description,
                "input_schema": tool.inputSchema,
            })
            server_for_tool[f"{server.name}_{tool.name}"] = (server, tool.name)

    # Claudeの呼び出し
    response = client.messages.create(
        tools=all_tools,
        messages=[{"role": "user", "content": user_message}],
    )

    # ツール呼び出し時、適切なサーバーにルーティング
    if response.stop_reason == "tool_use":
        for tu in response.content:
            if tu.type == "tool_use":
                server, original_name = server_for_tool[tu.name]
                result = await server.call_tool(original_name, tu.input)

HTTP トランスポートでの接続

# HTTP MCP サーバー（ローカル/リモート）

from mcp.client.session import ClientSession
from mcp.client.sse import sse_client

async def http_mcp_example():
    async with sse_client("https://mcp.example.com/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            # 以降は同じ

認証付きMCP サーバーの呼び出し

# Bearer トークン付きHTTP接続

server_params = StdioServerParameters(
    command="curl",
    args=[
        "-H", f"Authorization: Bearer {os.environ['MCP_TOKEN']}",
        "https://mcp.example.com",
    ],
)

# または専用クライアントで
async with sse_client(
    "https://mcp.example.com/sse",
    headers={"Authorization": f"Bearer {token}"},
) as (read, write):
    ...

このレッスンのまとめ

「Claude Desktop の設定ファイル」「Python/TypeScript SDK での直接呼び出し」「複数MCP統合」が主要な実装パターン。次のレッスンでは、デバッグとパフォーマンス計測を学びます。