生成AI入門 2026.05.11

MCPとは?Model Context Protocolを5分で理解【図解・実装コード付き 2026年版】

タグ:MCP / Claude / 初心者向け / 技術トレンド

MCPを3行で理解する

MCPなし: Claude → 人間が手でデータを貼り付け → 回答
MCP あり: Claude → 自動でDB/API/ファイルを検索 → 回答

MCPは「ClaudeやClaude Codeが外部のデータソース・ツールに自動接続するための共通規格」です。

仕組みの全体図

Claude Desktop / Claude Code
        ↓ stdio or HTTP
   MCPサーバー(あなたが作る or 公式)

  データベース / API / ファイルシステム

MCPサーバーは「使えるツール一覧(list_tools)」と「ツールの実行(call_tool)」の2つのエンドポイントを提供します。ClaudeはこれをAuto Discoveryして自動で使いこなします。

Claude Desktopへの登録方法

// ~/Library/Application Support/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "my-db-server": {
      "command": "python",
      "args": ["/Users/yourname/mcp-servers/database_server.py"],
      "env": {
        "DATABASE_URL": "postgresql://localhost:5432/myapp"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/yourname/projects"]
    }
  }
}

設定ファイルを保存してClaude Desktopを再起動すると、ClaudeがMCPサーバーを自動認識します。

PythonでカスタムMCPサーバーを作る

# weather_mcp_server.py
# pip install mcp httpx

import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

app = Server("weather-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    """Claudeに提供するツール一覧"""
    return [
        types.Tool(
            name="get_weather",
            description="指定した都市の現在の天気を取得する",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "都市名(例: Tokyo)"}
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """ツールを実行する"""
    if name == "get_weather":
        city = arguments["city"]
        async with httpx.AsyncClient() as client:
            # 実際はWeather APIを使う(例: wttr.inは無料)
            resp = await client.get(f"https://wttr.in/{city}?format=3")
            return [types.TextContent(type="text", text=resp.text)]
    
    return [types.TextContent(type="text", text=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())

SQLiteに接続するMCPサーバー

# sqlite_mcp_server.py
# 読み取り専用でSQLiteに接続するMCPサーバー

import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

DB_PATH = "./myapp.db"
app = Server("sqlite-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="query_db",
            description="SQLiteデータベースをSELECT文で検索する",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SELECT文のみ(INSERT/UPDATE/DELETE禁止)"}
                },
                "required": ["sql"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "query_db":
        sql = arguments["sql"].strip()
        
        # セキュリティ: SELECT以外を禁止
        if not sql.upper().startswith("SELECT"):
            return [types.TextContent(type="text", text="エラー: SELECT文のみ許可されています")]
        
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.execute(sql)
        rows = cursor.fetchall()
        cols = [d[0] for d in cursor.description]
        conn.close()
        
        result = f"カラム: {', '.join(cols)}\n"
        result += "\n".join(str(row) for row in rows[:50])  # 最大50行
        return [types.TextContent(type="text", text=result)]

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())

あわせて読みたい

参考ソース