MCP (Model Context Protocol) 入門

参照コース: https://anthropic.skilljar.com/introduction-to-model-context-protocol
対象読者: MCPを利用するだけでなく、サーバーを作る側の仕組みを学びたい人。

1. MCP とは何か

一言で: Claude（AIクライアント）と外部サービス（ツール・データ）をつなぐ標準プロトコル

Before MCP（ツールごとに実装が必要）:
Claude ──[独自コード]──> AWS API
Claude ──[独自コード]──> DB
Claude ──[独自コード]──> Git

After MCP（標準プロトコルで統一）:
Claude ──[MCP]──> MCP Server A (AWS tools)
Claude ──[MCP]──> MCP Server B (DB tools)
Claude ──[MCP]──> MCP Server C (Git tools)

claude-mem との関係: claude-mem 自体が MCP サーバー実装。インストールすると Claude が SQLite DB を検索できるようになる仕組みがまさにこれ。

2. MCP の3つのプリミティブ

3. Python で MCP サーバーを作る

pip install mcp

# server.py — AWS インフラ調査ツールを持つ MCP サーバー
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import boto3
import json

app = Server("aws-infra-tools")

# ── Tools ──────────────────────────────────────────────
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="list_ec2_instances",
            description="指定リージョンのEC2インスタンス一覧を取得する",
            inputSchema={
                "type": "object",
                "properties": {
                    "region": {
                        "type": "string",
                        "description": "AWSリージョン（例: ap-northeast-1）",
                        "default": "ap-northeast-1"
                    },
                    "state": {
                        "type": "string",
                        "enum": ["running", "stopped", "all"],
                        "default": "running"
                    }
                }
            }
        ),
        types.Tool(
            name="get_cloudwatch_metrics",
            description="EC2インスタンスのCPU使用率を取得する",
            inputSchema={
                "type": "object",
                "properties": {
                    "instance_id": {"type": "string"},
                    "hours": {"type": "integer", "default": 1}
                },
                "required": ["instance_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "list_ec2_instances":
        region = arguments.get("region", "ap-northeast-1")
        state = arguments.get("state", "running")
        
        ec2 = boto3.client('ec2', region_name=region)
        
        filters = [] if state == "all" else [
            {'Name': 'instance-state-name', 'Values': [state]}
        ]
        
        response = ec2.describe_instances(Filters=filters)
        instances = []
        for r in response['Reservations']:
            for inst in r['Instances']:
                name_tag = next(
                    (t['Value'] for t in inst.get('Tags', []) if t['Key'] == 'Name'),
                    'N/A'
                )
                instances.append({
                    'id': inst['InstanceId'],
                    'type': inst['InstanceType'],
                    'state': inst['State']['Name'],
                    'name': name_tag,
                    'private_ip': inst.get('PrivateIpAddress', 'N/A'),
                })
        
        return [types.TextContent(
            type="text",
            text=json.dumps(instances, ensure_ascii=False, indent=2)
        )]

# ── Resources ──────────────────────────────────────────
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="aws://ec2/regions",
            name="利用可能なAWSリージョン一覧",
            description="EC2が利用可能なリージョンの一覧",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "aws://ec2/regions":
        ec2 = boto3.client('ec2', region_name='ap-northeast-1')
        response = ec2.describe_regions()
        regions = [r['RegionName'] for r in response['Regions']]
        return json.dumps(regions)

# ── Prompts ─────────────────────────────────────────────
@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="investigate_instance",
            description="EC2インスタンスの障害調査を行う",
            arguments=[
                types.PromptArgument(
                    name="instance_id",
                    description="調査対象のインスタンスID",
                    required=True
                )
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
    if name == "investigate_instance":
        instance_id = arguments["instance_id"]
        return types.GetPromptResult(
            messages=[{
                "role": "user",
                "content": f"""
EC2インスタンス {instance_id} の障害調査をしてください。
以下のツールを使って情報を収集し、問題の特定と対処法を提示してください：
1. list_ec2_instances でインスタンス状態を確認
2. get_cloudwatch_metrics でCPU使用率を確認
3. 収集した情報から問題を分析
"""
            }]
        )

# エントリーポイント
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())

4. Claude Desktop への登録

// ~/AppData/Roaming/Claude/claude_desktop_config.json (Windows)
// ~/.config/claude/claude_desktop_config.json (Linux)
{
  "mcpServers": {
    "aws-infra-tools": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "AWS_PROFILE": "default",
        "AWS_DEFAULT_REGION": "ap-northeast-1"
      }
    }
  }
}

5. MCP Inspector でデバッグ

# ブラウザで動作確認できる
npx @modelcontextprotocol/inspector python server.py
# → http://localhost:5173 でツールの動作確認

6. 作ると役立つ MCP サーバーのアイデア

🔗 次のステップ

• 02_MCP応用.md — 参照コース: https://anthropic.skilljar.com/model-context-protocol-advanced-topics
• ../Claude/03_API活用.md — Claude API から MCP クライアントを実装