生成AI入門 2026.05.11
MCPとは?Model Context Protocolを5分で理解【図解・実装コード付き 2026年版】
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())