mcp-protocol-builder

MCP（Model Context Protocol） - AI ネイティブなサーバー開発

概要

Model Context Protocol（MCP）は、AI アシスタントを外部のデータソースやツールに接続するためのオープン標準です。ツール（LLM が呼び出せる関数）、リソース（LLM が読み取れるデータ）、プロンプト（LLM が使用できるテンプレート）を公開するサーバーを構築します。

主な概念：

• ツール：LLM が実行できる関数（ファイルの読み取り、API へのクエリ、コマンドの実行）
• リソース：LLM がアクセスできるデータソース（ファイル、データベース、API）
• プロンプト：一般的なタスク向けの引数付き再利用可能なテンプレート
• クライアント／サーバー：MCP サーバーが機能を公開し、クライアント（Claude Desktop など）がそれらを利用します
• トランスポート：STDIO（ローカル）、SSE（Server-Sent Events）、HTTP（ネットワーク）

公式 SDK：

• TypeScript：@modelcontextprotocol/sdk
• Python：mcp

インストール：

# TypeScript server
npx @modelcontextprotocol/create-server@latest my-server
cd my-server && npm install

# Python server
pip install mcp
# Or use uv (recommended)
uv pip install mcp

クイックスタート - TypeScript サーバー

1. CLI でサーバーを作成する

# Interactive setup
npx @modelcontextprotocol/create-server@latest my-filesystem-server

# Options prompt:
# - Server name: my-filesystem-server
# - Language: TypeScript
# - Include example tools: Yes

2. ツールを定義する

// src/index.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs/promises";
import * as path from "path";

const server = new Server(
  {
    name: "filesystem-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// List available tools
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "read_file",
        description: "Read contents of a file",
        inputSchema: {
          type: "object",
          properties: {
            path: {
              type: "string",
              description: "Path to the file to read",
            },
          },
          required: ["path"],
        },
      },
      {
        name: "list_directory",
        description: "List contents of a directory",
        inputSchema: {
          type: "object",
          properties: {
            path: {
              type: "string",
              description: "Directory path to list",
            },
          },
          required: ["path"],
        },
      },
    ],
  };
});

// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "read_file": {
      const filePath = args.path as string;
      const content = await fs.readFile(filePath, "utf-8");
      return {
        content: [{ type: "text", text: content }],
      };
    }

    case "list_directory": {
      const dirPath = args.path as string;
      const entries = await fs.readdir(dirPath, { withFileTypes: true });
      const listing = entries
        .map((entry) => `${entry.isDirectory() ? "📁" : "📄"} ${entry.name}`)
        .join("\n");
      return {
        content: [{ type: "text", text: listing }],
      };
    }

    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

// Start server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Filesystem MCP server running on stdio");
}

main();

3. Claude Desktop を設定する

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%/Claude/claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "filesystem": {
      "command": "node",
      "args": ["/absolute/path/to/my-filesystem-server/build/index.js"]
    }
  }
}

4. ビルドしてテストする

# Build TypeScript
npm run build

# Restart Claude Desktop (Cmd+Q and reopen)
# Server appears in 🔌 menu

クイックスタート - Python サーバー

1. サーバーを作成する

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

# Create server instance
app = Server("filesystem-server")

# Define tools
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="read_file",
            description="Read contents of a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to read"}
                },
                "required": ["path"],
            },
        ),
        Tool(
            name="list_directory",
            description="List directory contents",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Directory path"}
                },
                "required": ["path"],
            },
        ),
    ]

# Handle tool calls
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "read_file":
        file_path = arguments["path"]
        with open(file_path, "r") as f:
            content = f.read()
        return [TextContent(type="text", text=content)]

    elif name == "list_directory":
        dir_path = arguments["path"]
        entries = os.listdir(dir_path)
        listing = "\n".join(
            f"{'📁' if os.path.isdir(os.path.join(dir_path, e)) else '📄'} {e}"
            for e in entries
        )
        return [TextContent(type="text", text=listing)]

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

# Start server
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__":
    asyncio.run(main())

2. Claude Desktop を設定する

{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

3. テストする

# Test server standalone
python server.py

# Restart Claude Desktop
# Tools appear in 🔌 menu

リソース - LLM にデータを公開する

リソースを使うと、データソースへの読み取り専用アクセスを提供できます。

TypeScript のリソース例

import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// List available resources
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "file:///docs/readme.md",
        name: "README",
        description: "Project README documentation",
        mimeType: "text/markdown",
      },
      {
        uri: "file:///config/settings.json",
        name: "Settings",
        description: "Application settings",
        mimeType: "application/json",
      },
    ],
  };
});

// Handle resource reads
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const uri = request.params.uri;

  if (uri.startsWith("file://")) {
    const filePath = uri.replace("file://", "");
    const content = await fs.readFile(filePath, "utf-8");
    return {
      contents: [
        {
          uri,
          mimeType: "text/plain",
          text: content,
        },
      ],
    };
  }

  throw new Error(`Unknown resource: ${uri}`);
});

Python のリソース例

from mcp.types import Resource, ResourceContent

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="file:///docs/readme.md",
            name="README",
            description="Project README",
            mimeType="text/markdown",
        ),
        Resource(
            uri="file:///config/settings.json",
            name="Settings",
            description="App settings",
            mimeType="application/json",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> ResourceContent:
    if uri.startswith("file://"):
        file_path = uri.replace("file://", "")
        with open(file_path, "r") as f:
            content = f.read()
        return ResourceContent(uri=uri, mimeType="text/plain", text=content)

    raise ValueError(f"Unknown resource: {uri}")

プロンプト - 再利用可能なテンプレート

プロンプトは、LLM が引数とともに使用できるテンプレートです。

TypeScript のプロンプト例

import {
  ListPromptsRequestSchema,
  GetPromptRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// List prompts
server.setRequestHandler(ListPromptsRequestSchema, async () => {
  return {
    prompts: [
      {
        name: "code_review",
        description: "Review code for best practices",
        arguments: [
          {
            name: "language",
            description: "Programming language",
            required: true,
          },
          {
            name: "code",
            description: "Code to review",
            required: true,
          },
        ],
      },
    ],
  };
});

// Handle prompt requests
server.setRequestHandler(GetPromptRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "code_review") {
    const language = args?.language || "unknown";
    const code = args?.code || "";

    return {
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `Review this ${language} code for best practices, security issues, and improvements:\n\n\`\`\`${language}\n${code}\n\`\`\``,
          },
        },
      ],
    };
  }

  throw new Error(`Unknown prompt: ${name}`);
});

Python のプロンプト例

from mcp.types import Prompt, PromptMessage, PromptArgument

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code_review",
            description="Review code for best practices",
            arguments=[
                PromptArgument(
                    name="language", description="Programming language", required=True
                ),
                PromptArgument(name="code", description="Code to review", required=True),
            ],
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code_review":
        language = arguments.get("language", "unknown")
        code = arguments.get("code", "")

        return [
            PromptMessage(
                role="user",
                content={
                    "type": "text",
                    "text": f"Review this {language} code:\n\n```{language}\n{code}\n```",
                },
            )
        ]

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

高度なパターン

1. エラー処理

// TypeScript
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  try {
    const { name, arguments: args } = request.params;

    if (name === "risky_operation") {
      // Validate inputs
      if (!args.required_param) {
        throw new Error("Missing required parameter: required_param");
      }

      // Perform operation with proper error handling
      const result = await performRiskyOperation(args.required_param);

      return {
        content: [{ type: "text", text: JSON.stringify(result) }],
      };
    }
  } catch (error) {
    // Return error to LLM with helpful message
    return {
      content: [
        {
          type: "text",
          text: `Error: ${error instanceof Error ? error.message : String(error)}`,
        },
      ],
      isError: true,
    };
  }
});

# Python
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "risky_operation":
            if "required_param" not in arguments:
                raise ValueError("Missing required parameter: required_param")

            result = await perform_risky_operation(arguments["required_param"])
            return [TextContent(type="text", text=json.dumps(result))]

    except Exception as e:
        return [TextContent(type="text", text=f"Error: {str(e)}", isError=True)]

2. 非同期処理

// TypeScript - Async API calls
import axios from "axios";

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "fetch_api_data") {
    const url = args.url as string;
    const response = await axios.get(url);

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(response.data, null, 2),
        },
      ],
    };
  }
});

# Python - Async database queries
import aiosqlite

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "query_database":
        query = arguments["query"]

        async with aiosqlite.connect("database.db") as db:
            async with db.execute(query) as cursor:
                rows = await cursor.fetchall()
                result = json.dumps(rows)

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

3. 環境変数と設定

// TypeScript - Load config from environment
import dotenv from "dotenv";
dotenv.config();

const API_KEY = process.env.API_KEY;
const BASE_URL = process.env.BASE_URL || "https://api.example.com";

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "api_call") {
    const response = await fetch(`${BASE_URL}/endpoint`, {
      headers: { Authorization: `Bearer ${API_KEY}` },
    });
    // ...
  }
});

# Python - Load config from environment
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("API_KEY")
BASE_URL = os.getenv("BASE_URL", "https://api.example.com")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "api_call":
        headers = {"Authorization": f"Bearer {API_KEY}"}
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{BASE_URL}/endpoint", headers=headers) as resp:
                data = await resp.json()
                return [TextContent(type="text", text=json.dumps(data))]

4. ストリーミング応答（大容量データ）

// TypeScript - Stream large file contents
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "read_large_file") {
    const filePath = request.params.arguments.path as string;
    const stream = fs.createReadStream(filePath, { encoding: "utf-8" });

    let content = "";
    for await (const chunk of stream) {
      content += chunk;
      // Could yield chunks incrementally in future MCP versions
    }

    return {
      content: [{ type: "text", text: content }],
    };
  }
});

5. 動的なツール登録

// TypeScript - Register tools from config
interface ToolConfig {
  name: string;
  description: string;
  schema: object;
  handler: (args: any) => Promise<any>;
}

const toolRegistry = new Map<string, ToolConfig>();

function registerTool(config: ToolConfig) {
  toolRegistry.set(config.name, config);
}

// Register custom tools
registerTool({
  name: "custom_tool",
  description: "Dynamically registered tool",
  schema: {
    type: "object",
    properties: {
      input: { type: "string" },
    },
  },
  handler: async (args) => {
    return { result: `Processed: ${args.input}` };
  },
});

// List tools dynamically
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: Array.from(toolRegistry.values()).map((tool) => ({
      name: tool.name,
      description: tool.description,
      inputSchema: tool.schema,
    })),
  };
});

// Call tools dynamically
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const tool = toolRegistry.get(request.params.name);
  if (!tool) {
    throw new Error(`Unknown tool: ${request.params.name}`);
  }

  const result = await tool.handler(request.params.arguments);
  return {
    content: [{ type: "text", text: JSON.stringify(result) }],
  };
});

トランスポートの種類

1. STDIO（ローカル実行）

Claude Desktop との連携におけるデフォルトです。サーバーはサブプロセスとして実行されます。

// Claude Desktop config
{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["/path/to/server/build/index.js"],
      "env": {
        "API_KEY": "your-api-key"  # pragma: allowlist secret
      }
    }
  }
}

2. SSE（Server-Sent Events）

HTTP トランスポートを使用する長時間稼働サーバー向けです。

// TypeScript SSE server
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";

const app = express();
const server = new Server(/* ... */);

app.get("/sse", async (req, res) => {
  const transport = new SSEServerTransport("/messages", res);
  await server.connect(transport);
});

app.post("/messages", async (req, res) => {
  // Handle incoming messages
});

app.listen(3000, () => {
  console.log("MCP server listening on http://localhost:3000");
});

// Claude Desktop config for SSE
{
  "mcpServers": {
    "remote-server": {
      "url": "http://localhost:3000/sse"
    }
  }
}

3. HTTP（ストリーミング可能な HTTP）

REST スタイルの MCP サーバー向けです。

# Python HTTP server with FastAPI
from fastapi import FastAPI
from mcp.server.fastapi import create_fastapi_app

app = FastAPI()
mcp_app = Server("http-server")

# Define tools/resources/prompts as usual
# ...

# Mount MCP routes
app.mount("/mcp", create_fastapi_app(mcp_app))

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

MCP サーバーのデバッグ

1. サーバーログ

// TypeScript - Console.error for logs (STDOUT is for MCP protocol)
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  console.error(`Tool called: ${request.params.name}`);
  console.error(`Arguments: ${JSON.stringify(request.params.arguments)}`);
  // ...
});

# Python - Use logging module
import logging

logging.basicConfig(level=logging.DEBUG, filename="mcp-server.log")
logger = logging.getLogger(__name__)

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

2. MCP Inspector

# Install MCP Inspector (browser-based debugging tool)
npm install -g @modelcontextprotocol/inspector

# Run inspector with your server
mcp-inspector node /path/to/server/build/index.js

# Opens browser at http://localhost:6274
# - Test tools manually
# - See request/response payloads
# - Debug JSON schema validation

3. Claude Desktop のログ

# macOS
tail -f ~/Library/Logs/Claude/mcp*.log

# Windows
# Check %APPDATA%/Claude/logs/

4. サーバーを単独でテストする

// TypeScript - Add test harness
if (process.argv.includes("--test")) {
  // Simulate tool call
  const testRequest = {
    params: {
      name: "read_file",
      arguments: { path: "./test.txt" },
    },
  };

  server
    .setRequestHandler(CallToolRequestSchema, async (request) => {
      // ... your handler
    })
    .then((handler) => handler(testRequest))
    .then((result) => console.log(JSON.stringify(result, null, 2)))
    .catch((error) => console.error(error));
} else {
  // Normal server startup
  main();
}

# Test without Claude Desktop
node build/index.js --test

実際のサーバーの例

1. GitHub 連携サーバー

// github-server/src/index.ts
import { Octokit } from "@octokit/rest";

const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "list_repos",
        description: "List repositories for a user/org",
        inputSchema: {
          type: "object",
          properties: {
            owner: { type: "string", description: "Username or org name" },
          },
          required: ["owner"],
        },
      },
      {
        name: "create_issue",
        description: "Create a GitHub issue",
        inputSchema: {
          type: "object",
          properties: {
            owner: { type: "string" },
            repo: { type: "string" },
            title: { type: "string" },
            body: { type: "string" },
          },
          required: ["owner", "repo", "title"],
        },
      },
    ],
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "list_repos") {
    const { data } = await octokit.repos.listForUser({
      username: args.owner as string,
    });
    const repos = data.map((r) => ({
      name: r.name,
      description: r.description,
      stars: r.stargazers_count,
    }));
    return {
      content: [{ type: "text", text: JSON.stringify(repos, null, 2) }],
    };
  }

  if (name === "create_issue") {
    const { data } = await octokit.issues.create({
      owner: args.owner as string,
      repo: args.repo as string,
      title: args.title as string,
      body: args.body as string,
    });
    return {
      content: [
        { type: "text", text: `Issue created: ${data.html_url}` },
      ],
    };
  }
});

2. データベースクエリサーバー

# database-server/server.py
import asyncpg
import json
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("database-server")

async def get_db_pool():
    return await asyncpg.create_pool(
        host="localhost",
        database="mydb",
        user="user",
        password="password"  # pragma: allowlist secret
    )

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query",
            description="Execute SQL query (SELECT only)",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL query"},
                },
                "required": ["sql"],
            },
        ),
        Tool(
            name="list_tables",
            description="List all tables in database",
            inputSchema={"type": "object", "properties": {}},
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    pool = await get_db_pool()

    if name == "query":
        sql = arguments["sql"]
        # Security: Only allow SELECT
        if not sql.strip().upper().startswith("SELECT"):
            raise ValueError("Only SELECT queries allowed")

        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            result = [dict(row) for row in rows]

        return [TextContent(type="text", text=json.dumps(result, indent=2))]

    elif name == "list_tables":
        async with pool.acquire() as conn:
            tables = await conn.fetch("""
                SELECT table_name FROM information_schema.tables
                WHERE table_schema = 'public'
            """)
            result = [row["table_name"] for row in tables]

        return [TextContent(type="text", text=json.dumps(result, indent=2))]

3. ウェブスクレイパーサーバー

// scraper-server/src/index.ts
import * as cheerio from "cheerio";
import axios from "axios";

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "scrape_url",
        description: "Scrape content from a URL",
        inputSchema: {
          type: "object",
          properties: {
            url: { type: "string", description: "URL to scrape" },
            selector: {
              type: "string",
              description: "CSS selector for content",
            },
          },
          required: ["url"],
        },
      },
    ],
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "scrape_url") {
    const url = request.params.arguments.url as string;
    const selector = request.params.arguments.selector as string;

    const response = await axios.get(url);
    const $ = cheerio.load(response.data);

    let content: string;
    if (selector) {
      content = $(selector).text();
    } else {
      content = $("body").text();
    }

    return {
      content: [
        {
          type: "text",
          text: content.trim(),
        },
      ],
    };
  }
});

ベストプラクティス

1. ツールの設計

✅ 推奨事項：

• 明確で説明的なツール名を使用します（rf ではなく read_file）
• LLM が理解できるよう、詳細な説明を記述します
• 説明を含む包括的な JSON スキーマを定義します
• 可能な場合は構造化データ（JSON）を返します
• 役立つメッセージを添えて、適切にエラーを処理します

❌ 避けるべきこと：

• 範囲が広すぎるツールを作成しないでください（複雑な操作は分割します）
• 巨大なペイロードを返さないでください（大規模なデータセットにはページネーションを使用します）
• あいまいなパラメーター名を使用しないでください
• LLM がドメイン固有の用語を知っていると想定しないでください

2. セキュリティ

重要なルール：

• すべての入力を検証します（型、範囲、形式）
• ファイルパスをサニタイズします（ディレクトリトラバーサルを防止）
• コマンド／操作には許可リストを使用します
• 応答に機密性の高い認証情報を決して公開しません
• コストの高い操作にはレート制限を実装します
• デフォルトで読み取り専用アクセスを使用します

// Example: Path validation
function validatePath(inputPath: string): string {
  const normalized = path.normalize(inputPath);
  const allowed = path.resolve("/safe/directory");

  if (!normalized.startsWith(allowed)) {
    throw new Error("Path outside allowed directory");
  }

  return normalized;
}

3. パフォーマンス

• コストの高い操作をキャッシュします
• 可能な場合は大容量の応答をストリーミングします
• リスト操作にはページネーションを使用します
• 適切なタイムアウトを設定します
• レート制限のある API にはリクエストキューを実装します

4. テスト

# Use MCP Inspector for manual testing
mcp-inspector node build/index.js

# Unit test tool handlers
npm test

# Integration test with Claude Desktop
# 1. Add to config
# 2. Restart Claude
# 3. Test in conversation

5. ドキュメント

• すべてのツールについてコードコメントで説明します
• 説明に使用例を記載します
• ドキュメントにエラー発生時のケースを含めます
• サーバーの更新に合わせて CHANGELOG を管理します

よくある落とし穴

❌ STDOUT への書き込み（MCP プロトコルが破損します）：

// WRONG
console.log("Debug message");  // STDOUT is for MCP protocol

// CORRECT
console.error("Debug message");  // STDERR for logs

❌ エラーを処理しない：

// WRONG - unhandled promise rejection
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const data = await riskyOperation();  // Can throw
  return { content: [{ type: "text", text: data }] };
});

// CORRECT
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  try {
    const data = await riskyOperation();
    return { content: [{ type: "text", text: data }] };
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${error.message}` }],
      isError: true,
    };
  }
});

❌ ブロッキング処理：

# WRONG - synchronous file read blocks event loop
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    with open("large_file.txt", "r") as f:  # Blocks!
        content = f.read()
    return [TextContent(type="text", text=content)]

# CORRECT - async file I/O
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with aiofiles.open("large_file.txt", "r") as f:
        content = await f.read()
    return [TextContent(type="text", text=content)]

❌ スキーマに必須フィールドがない：

// WRONG - missing "required" field
{
  name: "search",
  inputSchema: {
    type: "object",
    properties: {
      query: { type: "string" }
    }
    // Missing: required: ["query"]
  }
}

リソース

• 公式ドキュメント: https://modelcontextprotocol.io
• TypeScript SDK: https://github.com/modelcontextprotocol/typescript-sdk
• Python SDK: https://github.com/modelcontextprotocol/python-sdk
• サーバーの例: https://github.com/modelcontextprotocol/servers
• MCP Inspector: https://github.com/modelcontextprotocol/inspector
• 仕様: https://spec.modelcontextprotocol.io

関連スキル

MCP サーバーを構築する際は、次の補完的なスキルを検討してください。

• typescript-core：TypeScript の型安全性、tsconfig の最適化、高度なパターン
• asyncio：async/await を使った MCP サーバー向け Python 非同期パターン
• openrouter：複数モデルをサポートする代替 LLM API 統合

TypeScript MCP パターンのクイックガイド（単独利用向けにインライン化）

// Type-safe MCP server with TypeScript
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';
import { z } from 'zod';

// Define tool schemas with runtime validation
const SearchSchema = z.object({
  query: z.string().min(1).max(500),
  limit: z.number().int().min(1).max(100).default(10),
});

type SearchArgs = z.infer<typeof SearchSchema>;

class TypeSafeMCPServer {
  private server: Server;

  constructor() {
    this.server = new Server(
      {
        name: 'typed-search-server',
        version: '1.0.0',
      },
      {
        capabilities: {
          tools: {},
        },
      }
    );

    this.setupToolHandlers();
  }

  private setupToolHandlers() {
    // List tools with full type inference
    this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
      tools: [
        {
          name: 'search',
          description: 'Search with type-safe parameters',
          inputSchema: {
            type: 'object',
            properties: {
              query: { type: 'string', minLength: 1, maxLength: 500 },
              limit: { type: 'number', minimum: 1, maximum: 100, default: 10 },
            },
            required: ['query'],
          },
        },
      ],
    }));

    // Type-safe tool execution
    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      if (request.params.name === 'search') {
        // Runtime validation with Zod
        const args = SearchSchema.parse(request.params.arguments);

        // Type-safe implementation
        const results = await this.performSearch(args);

        return {
          content: [{ type: 'text', text: JSON.stringify(results) }],
        };
      }

      throw new Error(`Unknown tool: ${request.params.name}`);
    });
  }

  private async performSearch(args: SearchArgs): Promise<Array<{ title: string; url: string }>> {
    // Implementation with full type safety
    // args.query is string, args.limit is number
    return [];
  }

  async run() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
  }
}

// Start server
const server = new TypeSafeMCPServer();
server.run();

Python Async MCP パターンのクイックガイド（単独利用向けにインライン化）

# Async MCP server with Python
import asyncio
import logging
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field

# Type-safe argument models with Pydantic
class SearchArgs(BaseModel):
    query: str = Field(..., min_length=1, max_length=500)
    limit: int = Field(10, ge=1, le=100)

class AsyncMCPServer:
    def __init__(self):
        self.server = Server("async-search-server")
        self.setup_handlers()

    def setup_handlers(self):
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="search",
                    description="Search with async processing",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "minLength": 1, "maxLength": 500},
                            "limit": {"type": "number", "minimum": 1, "maximum": 100, "default": 10},
                        },
                        "required": ["query"],
                    },
                )
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
            if name == "search":
                # Validate with Pydantic
                args = SearchArgs(**arguments)

                # Async implementation
                results = await self.perform_search(args)

                return [
                    TextContent(
                        type="text",
                        text=str(results)
                    )
                ]

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

    async def perform_search(self, args: SearchArgs) -> list[dict[str, str]]:
        """Async search implementation"""
        # Simulate async I/O
        await asyncio.sleep(0.1)

        # Use validated args (args.query is str, args.limit is int)
        return [
            {"title": f"Result for {args.query}", "url": "https://example.com"}
        ]

async def main():
    """Run async MCP server"""
    logging.basicConfig(level=logging.INFO)
    server = AsyncMCPServer()

    # Use stdio transport for Claude Desktop
    async with stdio_server() as (read_stream, write_stream):
        await server.server.run(
            read_stream,
            write_stream,
            server.server.create_initialization_options()
        )

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

多言語 MCP パターンのクイックガイド（単独利用向けにインライン化）

TypeScript と Python のトレードオフ：

機能	TypeScript	Python
型安全性	コンパイル時＋実行時（Zod）	実行時のみ（Pydantic）
パフォーマンス	起動が速い。Node.js のオーバーヘッドあり	起動が遅い。CPU タスクに適している
非同期処理のサポート	ネイティブの async/await、イベントループ	asyncio、I/O に最適
エコシステム	npm パッケージ、フロントエンドツール	データサイエンス、ML ライブラリ
最適な用途	Web API、リアルタイムツール	データ処理、ML 統合

両方に共通するパターン：

1. 入力の検証

   • TypeScript：Zod スキーマ
   • Python：Pydantic モデル

2. エラー処理

   • 両方：特定のエラー型を使った try/catch
   • MCP レスポンスでエラー内容を返す

3. リソース管理

   • TypeScript：try/finally を使った async/await
   • Python：非同期コンテキストマネージャー

4. テスト

   • TypeScript：モックトランスポートを使った Vitest/Jest
   • Python：pytest-asyncio を使った pytest

実装言語の選択：

// TypeScript - Best for:
// - File system operations
// - Web scraping/HTTP requests
// - JSON/API manipulation
// - Real-time data streams

// Python - Best for:
// - Data analysis (pandas, numpy)
// - Machine learning (scikit-learn, torch)
// - Database ETL operations
// - Scientific computing

[完全な TypeScript、Python async、OpenRouter のパターンは、同時にデプロイする場合、それぞれのスキルで利用できます]

まとめ

• MCP は、ツール、リソース、プロンプトを備えた AI ネイティブなサーバー開発を可能にします
• ツール は LLM が呼び出せる関数です（ファイルの読み取り、API のクエリ、操作の実行）
• リソース は読み取り専用のデータアクセスを提供します（ファイル、データベース、ドキュメント）
• プロンプト は、一般的なタスク向けの引数付き再利用可能なテンプレートです
• SDK は TypeScript と Python 向けに提供されており、完全な非同期処理をサポートします
• Claude Desktop は JSON 設定（STDIO トランスポート）を介して統合できます
• デバッグ には MCP Inspector、サーバーログ、単体テストを使用します
• セキュリティ は重要です。入力を検証し、パスをサニタイズし、許可リストを使用してください
• ベストプラクティス：明確な命名、包括的なスキーマ、エラー処理、パフォーマンスの最適化