Python MCP Server開発：2026年ゼロからAI Agentが呼び出せるツールサービスを構築する

AI Agentを開発していると、各AIプラットフォームが独自のツール呼び出しプロトコルを持っていることに気づいていませんか？Claudeは独自の方式、ChatGPTは別の方式、Geminiはまた違う方式を使っています。ツールコードを各プラットフォーム向けにアダプター層を書く必要がありますか？2026年、Model Context Protocol（MCP）はAIツール呼び出しのオープン標準となっています — MCP Serverを一度書けば、すべての主要AIプラットフォームから呼び出せます。

---

背景知識

MCPプロトコルアーキテクチャ

MCP（Model Context Protocol）は、Anthropicが2024年末にリリースしたオープンプロトコルで、AIアプリケーション（Host）とツールサービス（Server）間の標準的な通信方式を定義しています。

概念	説明	例え
Host	AIアプリケーション（Claude Desktop、ChatGPTなど）	ブラウザ
Client	Host内部のMCPクライアント	HTTP Client
Server	ツール/リソース/プロンプトを提供するサービス	HTTP Server
Transport	通信方式（stdio/SSE/Streamable HTTP）	TCP/HTTP

MCPコア機能

機能	説明	例
Tools	AIが呼び出し可能な関数	データベース照会、メール送信
Resources	AIが読み取り可能なデータ	ファイル内容、データベースレコード
Prompts	定義済みのプロンプトテンプレート	コードレビューテンプレート
Sampling	AIがHostに生成を要求	AIの判断が必要な中間ステップ

転送方式比較

転送	適用シーン	メリット	デメリット
stdio	ローカルデスクトップアプリ	シンプル、安全	ローカルプロセスのみ
SSE	リモートWebサービス	ネットワーク越え	長接続の維持
Streamable HTTP	リモートWebサービス（推奨）	ステートレス、CDNフレンドリー	MCP 2025-03仕様が必要

---

問題分析

なぜMCPが必要か？

1. プロトコルの断片化：各AIプラットフォームのツール呼び出し形式が異なり、適応コストが高い
2. ツールの再利用が困難：Claude向けに書いたツールをChatGPTで使用できない
3. セキュリティリスク：APIエンドポイントをAIに直接公開すると、権限制御が不十分
4. 状態管理の混乱：AIとツール間のコンテキスト渡しに標準がない

MCPは統一プロトコルでこれらの問題を解決します：一つのMCP Serverは、MCPをサポートする任意のAI Hostから呼び出せます。

---

ステップバイステップ実践

ステップ1：MCP Python SDKのインストール

pip install mcp[cli]

ステップ2：最初のMCP Serverを作成

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("toolsku-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",
            description="数式の計算を実行",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数式、例: '2 + 3 * 4'",
                    }
                },
                "required": ["expression"],
            },
        ),
        Tool(
            name="format_json",
            description="JSON文字列をフォーマット",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {
                        "type": "string",
                        "description": "フォーマットするJSON文字列",
                    },
                    "indent": {
                        "type": "integer",
                        "description": "インデントのスペース数、デフォルト2",
                        "default": 2,
                    },
                },
                "required": ["json_str"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "calculate":
        try:
            result = eval(arguments["expression"], {"__builtins__": {}}, {})
            return [TextContent(type="text", text=f"計算結果: {result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"計算エラー: {str(e)}")]

    elif name == "format_json":
        import json
        try:
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            formatted = json.dumps(parsed, indent=indent, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]
        except json.JSONDecodeError as e:
            return [TextContent(type="text", text=f"JSONパースエラー: {str(e)}")]

    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

ステップ3：Claude Desktop接続の設定

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "toolsku-tools": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "E:\\project\\toolsku\\mcp-server"
    }
  }
}

ステップ4：SSE転送モード（リモートデプロイ）

# sse_server.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route

app = Server("toolsku-remote-tools")
sse = SseServerTransport("/messages/")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="技術ドキュメントを検索",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "検索キーワード"},
                    "category": {"type": "string", "description": "ドキュメントカテゴリ"},
                    "limit": {"type": "integer", "description": "返却数", "default": 5},
                },
                "required": ["query"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = await search_documents(
            query=arguments["query"],
            category=arguments.get("category"),
            limit=arguments.get("limit", 5),
        )
        import json
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
    raise ValueError(f"Unknown tool: {name}")

async def search_documents(query: str, category: str | None, limit: int) -> list[dict]:
    return [
        {"title": f"ドキュメント: {query}", "url": f"https://docs.example.com/{query}", "score": 0.95},
    ][:limit]

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

ステップ5：ChatGPTとの統合

# chatgpt_adapter.py
import httpx
from openai import OpenAI

MCP_SERVER_URL = "http://localhost:8080"

async def discover_mcp_tools() -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/list",
                "id": 1,
            },
        )
        result = response.json()
        return result.get("result", {}).get("tools", [])

def mcp_tools_to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
    openai_tools = []
    for tool in mcp_tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool["name"],
                "description": tool["description"],
                "parameters": tool["inputSchema"],
            },
        })
    return openai_tools

async def call_mcp_tool(name: str, arguments: dict) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {"name": name, "arguments": arguments},
                "id": 2,
            },
        )
        result = response.json()
        contents = result.get("result", {}).get("content", [])
        return "\n".join(c.get("text", "") for c in contents)

async def chat_with_tools(user_message: str) -> str:
    mcp_tools = await discover_mcp_tools()
    openai_tools = mcp_tools_to_openai_tools(mcp_tools)

    client = OpenAI()
    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=openai_tools,
    )

    message = response.choices[0].message

    if message.tool_calls:
        for tool_call in message.tool_calls:
            tool_result = await call_mcp_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments),
            )
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result,
            })

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=openai_tools,
        )
        return response.choices[0].message.content

    return message.content

---

完全なコード：本番グレードMCP Server

# production_server.py
import os
import json
import logging
from datetime import datetime
from typing import Any

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent, ImageContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("toolsku-mcp")

app = Server("toolsku-production")

API_KEY = os.environ.get("MCP_API_KEY", "dev-key")

def validate_api_key(headers: dict) -> bool:
    auth = headers.get("authorization", "")
    return auth == f"Bearer {API_KEY}"

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="base64_encode",
            description="文字列をBase64エンコード",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "エンコードするテキスト"},
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="base64_decode",
            description="Base64文字列をデコード",
            inputSchema={
                "type": "object",
                "properties": {
                    "encoded": {"type": "string", "description": "Base64エンコードされた文字列"},
                },
                "required": ["encoded"],
            },
        ),
        Tool(
            name="json_format",
            description="JSON文字列をフォーマット",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {"type": "string", "description": "JSON文字列"},
                    "indent": {"type": "integer", "description": "インデント", "default": 2},
                    "sort_keys": {"type": "boolean", "description": "キーをソートするか", "default": False},
                },
                "required": ["json_str"],
            },
        ),
        Tool(
            name="hash_compute",
            description="テキストのハッシュ値を計算",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "ハッシュ計算するテキスト"},
                    "algorithm": {
                        "type": "string",
                        "enum": ["md5", "sha1", "sha256", "sha512"],
                        "description": "ハッシュアルゴリズム",
                        "default": "sha256",
                    },
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="timestamp_convert",
            description="タイムスタンプと日付形式の相互変換",
            inputSchema={
                "type": "object",
                "properties": {
                    "value": {"type": "string", "description": "タイムスタンプまたは日付文字列"},
                    "format": {"type": "string", "description": "ターゲット形式", "default": "iso"},
                },
                "required": ["value"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.info(f"Tool called: {name} with args: {arguments}")

    try:
        if name == "base64_encode":
            import base64
            result = base64.b64encode(arguments["text"].encode()).decode()
            return [TextContent(type="text", text=result)]

        elif name == "base64_decode":
            import base64
            result = base64.b64decode(arguments["encoded"]).decode()
            return [TextContent(type="text", text=result)]

        elif name == "json_format":
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            sort_keys = arguments.get("sort_keys", False)
            formatted = json.dumps(parsed, indent=indent, sort_keys=sort_keys, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]

        elif name == "hash_compute":
            import hashlib
            algorithm = arguments.get("algorithm", "sha256")
            h = hashlib.new(algorithm)
            h.update(arguments["text"].encode())
            return [TextContent(type="text", text=h.hexdigest())]

        elif name == "timestamp_convert":
            value = arguments["value"]
            fmt = arguments.get("format", "iso")
            try:
                ts = float(value)
                dt = datetime.fromtimestamp(ts)
            except ValueError:
                dt = datetime.fromisoformat(value)
                ts = dt.timestamp()

            if fmt == "iso":
                result = dt.isoformat()
            elif fmt == "timestamp":
                result = str(ts)
            else:
                result = dt.strftime(fmt)

            return [TextContent(type="text", text=result)]

        raise ValueError(f"Unknown tool: {name}")

    except Exception as e:
        logger.error(f"Tool error: {name} - {str(e)}")
        return [TextContent(type="text", text=f"エラー: {str(e)}")]

sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
]

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
    middleware=middleware,
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

---

落とし穴ガイド

落とし穴1：Claude DesktopがMCP Serverを発見できない

現象：claude_desktop_config.jsonを設定したが、Claude Desktopにツールが表示されない。

解決：JSON形式が正しいか確認（カンマと引用符に注意）、commandパスが正しいことを確認（絶対パスを使用）、Claude Desktopのログを確認（Help > Toggle Developer Tools）。

落とし穴2：SSE接続が頻繁に切断される

現象：リモートMCP ServerのSSE接続が30秒ごとに切断される。

解決：サーバー側にハートビート機構を実装（15秒ごとにkeep-aliveを送信）、リバースプロキシのタイムアウトを60秒以上に設定。Nginxではproxy_read_timeout 300sが必要。

落とし穴3：ツール引数の検証が甘くランタイムエラーが発生

現象：AIが間違った型の引数を渡す（例：整数が必要な場所に文字列を渡す）。

解決：inputSchemaで型と制約を厳密に定義（type、minimum、enumなど）、call_tool関数で引数検証と型変換を追加。

落とし穴4：stdioモードでPython環境が不一致

現象：Claude DesktopがMCP Serverを起動する際、インストール済みパッケージが見つからない。

解決：claude_desktop_config.jsonでPythonインタープリタの絶対パスを使用するか、uv runで仮想環境を指定：

{
  "mcpServers": {
    "toolsku-tools": {
      "command": "uv",
      "args": ["run", "--directory", "E:\\project\\toolsku\\mcp-server", "server.py"]
    }
  }
}

落とし穴5：並行リクエストで状態が混乱

現象：複数のAIが同じMCP Serverを同時に呼び出し、共有状態が上書きされる。

解決：MCP Serverはステートレスに設計し、すべての状態をパラメータで渡す。状態が必要な場合は、リクエストレベルのコンテキスト分離を使用し、グローバル変数を避ける。

---

エラートラブルシューティング

番号	エラーメッセージ	原因	解決方法
1	MCP server not found in config	設定ファイルのパスまたは形式エラー	JSON形式を確認、絶対パスを使用
2	Connection refused on stdio	Pythonプロセスの起動失敗	commandパスを確認、stderrログを確認
3	SSE connection timeout	サーバー未応答またはネットワーク問題	サーバー状態を確認、タイムアウトを増加
4	Tool not found: xxx	ツール名のタイプミス	list_toolsとcall_toolの名前が一致するか確認
5	JSON-RPC error: -32600	リクエスト形式がJSON-RPC仕様に非準拠	リクエストbody形式を確認
6	Permission denied	ツール実行に権限が必要	ファイル/ネットワーク権限を確認、sudoは慎重に使用
7	ModuleNotFoundError	Python依存パッケージが未インストール	MCP設定で正しいPython環境を指定
8	413 Payload Too Large	リクエストbodyが大きすぎる	ツール引数サイズを制限、ページネーションを使用
9	CORS error	SSEクロスオリジンリクエストがブロック	CORSミドルウェアを設定
10	Server initialization failed	MCPハンドシェイク失敗	initializeレスポンス形式とcapabilitiesを確認

---

高度な最適化

1. リソース（Resources）の公開

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="toolsku:///docs/api-reference",
            name="APIリファレンスドキュメント",
            mimeType="text/markdown",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "toolsku:///docs/api-reference":
        return "# APIリファレンス\n\n## ツール一覧\n..."
    raise ValueError(f"Unknown resource: {uri}")

2. プロンプトテンプレート（Prompts）

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code-review",
            description="コードレビュープロンプトテンプレート",
            arguments=[
                PromptArgument(name="language", description="プログラミング言語", required=True),
                PromptArgument(name="code", description="レビュー対象のコード", required=True),
            ],
        ),
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code-review":
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"以下の{arguments['language']}コードをレビューしてください。セキュリティ、パフォーマンス、保守性に注目：\n\n```{arguments['language']}\n{arguments['code']}\n```",
                ),
            ),
        ]
    raise ValueError(f"Unknown prompt: {name}")

3. Streamable HTTP転送（MCP 2025-03）

from mcp.server.streamable_http import StreamableHTTPTransport

http_transport = StreamableHTTPTransport("/mcp")

async def handle_http(request):
    await http_transport.handle_request(request, app)

4. 認証とレート制限

from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path in ["/sse", "/messages/"]:
            auth = request.headers.get("authorization", "")
            if auth != f"Bearer {API_KEY}":
                from starlette.responses import JSONResponse
                return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

---

比較分析

観点	MCP	OpenAI Function Calling	LangChain Tools	LlamaIndex Tools
プロトコル標準	オープン標準	プラットフォーム独自	フレームワーク独自	フレームワーク独自
クロスプラットフォーム	MCP対応の全Host	OpenAIのみ	LangChainのみ	LlamaIndexのみ
転送方式	stdio/SSE/HTTP	HTTP API	Python関数	Python関数
ツール発見	動的list_tools	静的定義	静的定義	静的定義
リソース公開	ネイティブサポート	非サポート	非サポート	非サポート
セキュリティモデル	権限制御	API Key	標準なし	標準なし
デプロイ方式	ローカル/リモート	リモート	プロセス内	プロセス内
エコシステム	急速成長中	成熟	成熟	成熟

---

まとめと展望

まとめ：MCPは2026年にAIツール呼び出しのオープン標準となりました。Python MCP SDKにより、MCP Serverの開発は極めてシンプルになります — ツールを定義し、ハンドラを登録し、転送方式を選択するだけです。コアの利点は「一度開発、マルチプラットフォーム呼び出し」：同じMCP Serverは、Claude Desktop、ChatGPT、Cursorなど、MCPをサポートする任意のAIアプリケーションから使用できます。シンプルなツールサービス（JSONフォーマット、Base64エンコード/デコードなど）から始め、段階的にリソースとプロンプトテンプレート機能を追加し、最終的に完全なAIツールエコシステムを構築することをお勧めします。

---

オンラインツール推薦

• JSONフォーマッター（MCPツール開発デバッグ）：/ja/json/format
• Base64エンコーダ/デコーダ（MCPツール機能）：/ja/encode/base64
• curl to Code（MCP APIテスト）：/ja/dev/curl-to-code