Claude CodeとMCP Serverによる次世代AIエージェント開発ガイド

はじめに

現代のソフトウェア開発現場において、AIエージェントは単なるチャットボットから、コード生成やデバッグを行うパートナーへと進化しています。しかし、AIエージェントが真の生産性向上をもたらすためには、社内のデータベース、API、ファイルシステムといった「外部リソース」へ安全かつ柔軟にアクセスする能力が不可欠です。従来のカスタムAPIラッパーやFunction Callingの実装では、認証管理やコンテキストの切り替え、プロトコルの標準化において多くの課題が存在しました。

本記事では、Anthropicが提唱するMCP (Model Context Protocol) と、それを実行する環境であるClaude Codeに焦点を当てます。MCPを利用することで、AIエージェントと開発ツール間の標準化されたインターフェースを構築でき、開発者は複雑な連携ロジックから解放されます。ここでは、MCPの内部構造から、TypeScriptおよびPythonによる実践的な実装例、そして実際のビジネスシーンでの活用方法まで、エンジニア向けに深く掘り下げて解説します。

MCP (Model Context Protocol) とは？

MCPは、AIアシスタント（クライアント）と外部ツールやデータソース（サーバー）との間で標準化された通信を行うためのオープンプロトコルです。従来、AIに特定の機能を追加するには、各AIベンダーが提供する独自のプラグイン形式やAPI仕様に従う必要がありましたが、MCPはベンダーに依存しない共通の規格を提供します。

MCPサーバーは「ツール（Tools）」「リソース（Resources）」「プロンプト（Prompts）」という3つの主要なリソースをクライアントに公開します。ツールは関数実行、リソースはデータの読み取り（ファイルやDBレコード）、プロンプトはテンプレート的なテキスト生成に対応します。このアーキテクチャにより、Claude Codeのようなクライアントは、サーバーの実装詳細を知らなくても、標準的なJSON-RPCメッセージを介してこれらの機能を呼び出せます。

既存のカスタムAPI連携と比較した場合、MCPの最大の利点は「トランスポートの抽象化」です。ローカル開発環境では標準入出力（stdio）を、クラウド環境ではSSE（Server-Sent Events）など、環境に合わせて通信手段を切り替え可能です。これにより、開発者はビジネスロジックに集中でき、インフラストラクチャの差異を吸収できます。

MCP Serverの内部動作とアーキテクチャ

MCP ServerはJSON-RPC 2.0プロトコルに基づいて動作します。クライアントからのリクエスト（methods/call）を受け取り、定義されたハンドラーを実行し、その結果をJSON形式で返すというシンプルな流れです。しかし、実運用においては、リクエストごとの認証、エラーハンドリング、ログの追跡など、考慮すべき点は多岐にわたります。

特に重要なのが「コンテキストの管理」です。AIエージェントは同時に複数のタスクを並列で実行することがあります。Node.js環境では AsyncLocalStorage を活用することで、非同期処理の深い階層においても、リクエストIDやユーザー権限などのコンテキスト情報を透過的に受け渡すことが可能です。これにより、どのツール実行がどのユーザーリクエストに起因したかを特定しやすくなり、デバッグや監査が容易になります。

また、データベースアクセスを行うMCP Serverでは、ORMの最新機能を活用することが推奨されます。Prismaであれば Client Extensions を、SQLAlchemyであれば2.0以降の select 文や do_orm_execute イベントを利用することで、型安全性を保ちつつ、クエリの最適化や取得データのフォーマット変換を効率的に行えます。

実装例：TypeScriptによるファイル監査MCP Server

最初の実装例では、Node.jsとTypeScriptを使用し、プロジェクト内のソースコードを静的解析するMCP Serverを構築します。このサーバーは AsyncLocalStorage を用いてリクエストトレースを行い、指定されたファイルパスの構文チェックや特定のパターン検索を行います。

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  Tool,
} from "@modelcontextprotocol/sdk/types.js";
import { AsyncLocalStorage } from "async_hooks";
import { readFile, access } from "fs/promises";
import * as path from "path";
import * as zod from "zod";

// リクエストコンテキストの型定義
interface RequestContext {
  requestId: string;
  timestamp: number;
}

// AsyncLocalStorageのインスタンス化（リクエストスコープのデータ管理）
const requestContextStorage = new AsyncLocalStorage<RequestContext>();

// ロギングユーティリティ（コンテキスト情報を含めて出力）
const logger = {
  info: (message: string, data?: unknown) => {
    const store = requestContextStorage.getStore();
    const prefix = store ? `[ReqID: ${store.requestId}]` : "[System]";
    console.log(JSON.stringify({ level: "info", context: prefix, message, data }));
  },
  error: (message: string, error?: unknown) => {
    const store = requestContextStorage.getStore();
    const prefix = store ? `[ReqID: ${store.requestId}]` : "[System]";
    console.error(JSON.stringify({ level: "error", context: prefix, message, error }));
  },
};

// ツール定義：ファイルの内容と行数を取得する
const ANALYZE_FILE_TOOL: Tool = {
  name: "analyze_file",
  description: "指定されたファイルの内容を読み込み、簡易的な統計情報（行数、文字数）を返します。",
  inputSchema: {
    type: "object",
    properties: {
      file_path: {
        type: "string",
        description: "解析対象のファイルパス（プロジェクトルートからの相対パス）",
      },
    },
    required: ["file_path"],
  },
};

// Zodスキーマによるバリデーション定義
const AnalyzeFileArgsSchema = zod.object({
  file_path: zod.string().min(1, "ファイルパスは必須です"),
});

// MCPサーバーの初期化
const server = new Server(
  {
    name: "project-auditor-mcp-server",
    version: "0.1.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// 利用可能なツールの一覧を返すハンドラー
server.setRequestHandler(ListToolsRequestSchema, async () => {
  logger.info("Tools list requested");
  return {
    tools: [ANALYZE_FILE_TOOL],
  };
});

// ツール実行ハンドラー
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  // 新しいコンテキストを作成して非同期ストレージに設定
  return await requestContextStorage.run(
    { requestId: crypto.randomUUID(), timestamp: Date.now() },
    async () => {
      const { name, arguments: args } = request.params;

      logger.info(`Tool execution started: ${name}`, { args });

      if (name === "analyze_file") {
        // バリデーション実行
        const validation = AnalyzeFileArgsSchema.safeParse(args);
        if (!validation.success) {
          logger.error("Validation failed", validation.error.flatten());
          throw new Error(`引数エラー: ${JSON.stringify(validation.error.flatten())}`);
        }

        const { file_path } = validation.data;
        const resolvedPath = path.resolve(process.cwd(), file_path);

        try {
          // ファイルの存在確認
          await access(resolvedPath);
          const content = await readFile(resolvedPath, "utf-8");
          const lines = content.split("\n").length;
          
          logger.info("File analyzed successfully", { path: resolvedPath, lines });

          return {
            content: [
              {
                type: "text",
                text: JSON.stringify({
                  status: "success",
                  path: resolvedPath,
                  stats: {
                    lines,
                    characters: content.length,
                    size_bytes: Buffer.byteLength(content, "utf-8"),
                  },
                  preview: content.substring(0, 200) + (content.length > 200 ? "..." : ""),
                }, null, 2),
              },
            ],
          };
        } catch (err: any) {
          logger.error("Error processing file", err);
          return {
            content: [
              {
                type: "text",
                text: JSON.stringify({
                  status: "error",
                  message: `ファイル処理中にエラーが発生しました: ${err.message}`,
                }),
              },
            ],
            isError: true,
          };
        }
      }

      logger.error(`Unknown tool requested: ${name}`);
      throw new Error(`不明なツールが呼び出されました: ${name}`);
    }
  );
});

// サーバー起動処理
async function main() {
  logger.info("Starting MCP Server...");
  const transport = new StdioServerTransport();
  await server.connect(transport);
  logger.info("MCP Server running on stdio");
}

main().catch((error) => {
  logger.error("Fatal error in main", error);
  process.exit(1);
});

このコードでは、@modelcontextprotocol/sdk を使用してサーバーを構築しています。AsyncLocalStorage を導入することで、並列に実行される可能性のあるツール呼び出しに対して、それぞれ一意な requestId を紐付け、ログの追跡性を確保しています。また、Zodを用いた入力バリデーションを行うことで、不正な引数による実行時エラーを未然に防ぎ、堅牢性を高めています。

実装例：PythonとSQLAlchemy 2.0によるDB連携サーバー

次に、Pythonを使用してデータベースと連携するMCP Serverの例を示します。ここでは、SQLAlchemy 2.0以降のスタイル（1.xのようなクエリ形式ではなく、select() を使用したスタイル）を採用し、顧客データを検索・取得するAPIを提供します。エラーハンドリングと接続管理にも注意を払っています。

import asyncio
import json
import logging
from contextlib import asynccontextmanager
from typing import Any, Sequence

from mcp.server.fastapi import FastAPI as MCPServer
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field, ValidationError
from sqlalchemy import select, String
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# ロギングの設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-db-server")

# データベースモデル定義 (SQLAlchemy 2.0記法)
class Base(DeclarativeBase):
    pass

class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255))
    tier: Mapped[str] = mapped_column(String(20))  # e.g., 'free', 'pro', 'enterprise'

# 非同期DBエンジン設定 (SQLiteを使用した例)
DATABASE_URL = "sqlite+aiosqlite:///:memory:"
engine = create_async_engine(DATABASE_URL, echo=False)
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)

# ツール引数のスキーマ定義 (Pydantic)
class FetchCustomerArgs(BaseModel):
    customer_id: int = Field(..., description="取得する顧客のID")
    include_history: bool = Field(default=False, description="購入履歴を含めるかどうか")

# MCPツールの定義
FETCH_CUSTOMER_TOOL = Tool(
    name="fetch_customer",
    description="指定されたIDの顧客情報をデータベースから取得します。",
    inputSchema={
        "type": "object",
        "properties": {
            "customer_id": {"type": "integer", "description": "顧客ID"},
            "include_history": {"type": "boolean", "description": "履歴を含むか", "default": False}
        },
        "required": ["customer_id"]
    }
)

# データベースセッション管理コンテキスト
@asynccontextmanager
async def get_db_session():
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# サーバーのセットアップ
mcp = MCPServer("mcp-db-server")

@mcp.list_tools()
async def list_tools() -> list[Tool]:
    return [FETCH_CUSTOMER_TOOL]

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

    if name == "fetch_customer":
        try:
            # Pydanticによるバリデーション
            args = FetchCustomerArgs(**arguments)
        except ValidationError as e:
            logger.error(f"Validation error: {e}")
            return [TextContent(
                type="text",
                text=json.dumps({"error": "Invalid arguments", "details": e.errors()})
            )]

        try:
            async with get_db_session() as session:
                # SQLAlchemy 2.0のselect文を使用
                stmt = select(Customer).where(Customer.id == args.customer_id)
                result = await session.execute(stmt)
                customer = result.scalar_one_or_none()

                if not customer:
                    return [TextContent(
                        type="text",
                        text=json.dumps({"error": "Customer not found", "id": args.customer_id})
                    )]

                # 結果を辞書形式に変換
                response_data = {
                    "id": customer.id,
                    "name": customer.name,
                    "email": customer.email,
                    "tier": customer.tier,
                    "include_history_flag": args.include_history
                }
                
                logger.info(f"Successfully fetched customer {customer.id}")
                return [TextContent(type="text", text=json.dumps(response_data, indent=2))]

        except Exception as e:
            logger.exception("Database error occurred")
            return [TextContent(
                type="text",
                text=json.dumps({"error": "Internal Server Error", "message": str(e)})
            )]

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

# 初期化時にダミーデータを作成
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    async with async_session_factory() as session:
        dummy_customer = Customer(name="Test User", email="test@example.com", tier="pro")
        session.add(dummy_customer)
        await session.commit()
    logger.info("Database initialized with dummy data.")

# エントリーポイント
if __name__ == "__main__":
    import uvicorn
    
    # 初期化実行
    asyncio.run(init_db())
    
    # サーバー起動 (stdio transportを使用)
    # FastAPIベースのMCPサーバーでは、uvicorn経由でstdioをハンドルするか、
    # SDKが提供するstdioトランスポートアダプタを使用します。
    # ここでは概念を示すため、簡易的な起動コマンドを記述します。
    # 実際には mcp.server.stdio を使用して実行することが一般的です。
    logger.info("Starting DB MCP Server...")
    # 実運用では mcp.server.stdio.serve() 等を使用

このPython実装では、SQLAlchemyの非機能要件である非同期I/O (AsyncSession) を最大限に活用しています。select(Customer).where(...) という2.0スタイルのクエリを使用することで、型ヒントが効いた安全なデータアクセスを実現しています。また、Pydanticを用いてリクエスト引数のバリデーションを行い、データベースアクセス前に不正なリクエストを弾く設計としています。

実装例：Prisma Client Extensionsを活用した高度なデータ処理

最後に、TypeScriptとPrismaを使用した、より高度な実装例を紹介します。ここでは、Prisma Client Extensions ($extends) を使用して、データベースから取得した生のデータを、AIが解釈しやすいフォーマットに自動変換するロジックを組み込みます。これにより、ツール実行関数内の変換ロジックを削減し、再利用性を高めます。

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 { PrismaClient } from "@prisma/client";

// Prisma Clientの初期化
const prisma = new PrismaClient();

// Prisma Client Extensionsの定義
// ユーザー情報をAI用に最適化されたサマリー形式に変換する拡張
const extendedPrisma = prisma.$extends({
  result: {
    user: {
      // 新しいフィールド 'aiSummary' を追加
      aiSummary: {
        needs: { id: true, email: true, role: true, createdAt: true },
        compute(user) {
          // ロジック: ユーザー情報を自然言語のような要約文字列に変換
          const daysSinceCreation = Math.floor(
            (Date.now() - user.createdAt.getTime()) / (1000 * 60 * 60 * 24)
          );
          return `User ID ${user.id} (${user.email}) is a ${user.role} member for ${daysSinceCreation} days.`;
        },
      },
    },
  },
});

// ツール定義
const GET_USER_SUMMARY_TOOL = {
  name: "get_user_summary",
  description: "ユーザーIDに基づき、AI分析用に最適化されたユーザーサマリーを取得します。",
  inputSchema: {
    type: "object",
    properties: {
      user_id: { type: "number", description: "対象のユーザーID" },
    },
    required: ["user_id"],
  },
};

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

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [GET_USER_SUMMARY_TOOL],
}));

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

  if (name === "get_user_summary") {
    const userId = args?.user_id as number;

    if (typeof userId !== "number") {
      return {
        content: [{ type: "text", text: "Error: user_id must be a number" }],
        isError: true,
      };
    }

    try {
      // 拡張されたClientを使用してデータを取得
      // 'aiSummary' フィールドが自動的に計算される
      const user = await extendedPrisma.user.findUnique({
        where: { id: userId },
        select: {
          id: true,
          email: true,
          aiSummary: true, // 拡張フィールドをselect
        },
      });

      if (!user) {
        return {
          content: [{ type: "text", text: `User with ID ${userId} not found.` }],
        };
      }

      // AIにとって処理しやすいJSON構造を返却
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(
              {
                id: user.id,
                email: user.email,
                analysis_context: user.aiSummary, // 拡張機能で生成された要約
              },
              null,
              2
            ),
          },
        ],
      };
    } catch (error) {
      console.error("Prisma Error:", error);
      return {
        content: [
          { type: "text", text: "Database connection error or query failed." },
        ],
        isError: true,
      };
    }
  }

  return {
    content: [{ type: "text", text: `Unknown tool: ${name}` }],
    isError: true,
  };
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Prisma MCP Server running...");
}

main();

このコードのポイントは prisma.$extends です。データベースから取得した生のデータ（createdAt などの日付型）を、そのままAIに渡しても文脈を理解するのは難しい場合があります。拡張機能内で「アカウント保有期間」を計算し、自然言語に近いサマリー文字列を生成することで、ClaudeなどのLLMがより高度な推論を行えるようになります。これは、データ層とAI利用層の間に「意味変換レイヤー」を設けるアーキテクチャの好例です。

技術的な選択肢：MCP vs 従来手法

MCPを導入する際、既存のカスタムAPI連携や他のプロトコルと比較して、どのようなメリットとデメリットがあるかを理解することが重要です。以下の表は、MCPと従来のアプローチを比較したものです。

アプローチ	メリット	デメリット	適したケース
MCP (Model Context Protocol)	ベンダー非依存の標準化された仕様。stdio/SSEなど柔軟なトランスポート。ツールの定義が自己記述的。	エコシステムが比較的新しいため、情報が少ない場合がある。すべてのAIクライアントが対応しているわけではない。	Claude DesktopやMCP対応エディタと連携し、ローカル開発環境を強化したい場合。
カスタム HTTP API + Function Calling	既存のWeb API知識が活かせる。HTTPサーバーさえあれば即座に実装可能。	認証やエラーハンドリングを独自に実装する必要がある。AIごとにスキーマ定義が異なる可能性がある。	クラウド上のAIサービスと連携し、既存のバックエンドAPIをそのまま利用したい場合。
LangChain Tools / Plugins	豊富な既存ツールセット。Python/JSエコシステムとの親和性が高い。	フレームワークへの依存度が高い。特定のLLMプロバイダーにロックインされやすい。	複雑なエージェントワークフローを構築しており、高度な制御が必要な場合。

MCPを選択する最大の理由は「開発体験の向上」と「将来の互換性」です。特にローカル開発環境において、AIエージェントにファイルシステムやGitリポジトリへのアクセスを許可する場合、MCPのセキュリティモデルと標準化されたインターフェースは非常に強力です。

ビジネスユースケース：カスタマーサポート自動化

MCP Serverの技術的な利点を具体的なビジネスシーンに当てはめてみましょう。あるECサイトを運営する企業を想定します。

課題: カスタマーサポートチームは、顧客からの問い合わせに対し、注文履歴、在庫状況、配送状況を別々のシステム（CRM、在庫管理DB、配送API）から確認する必要があり、回答に時間がかかっていました。

解決策: Claude CodeとMCP Serverを活用し、これら3つのシステムにアクセスするための統合エージェントを構築しました。

1. CRM MCP Server: 顧客の基本情報と過去の問い合わせ履歴を取得（PostgreSQL + Prisma実装）。
2. Inventory MCP Server: 現在の在庫状況を確認（Python + SQLAlchemy実装）。
3. Shipping MCP Server: 配送ステータスを追跡（外部APIラッパー）。

効果: サポート担当者はチャットボックス（Claude）に「顧客ID 12345の最新の注文状況と、商品Aの在庫はあるか？」と尋ねるだけで、エージェントがバックグラウンドで各MCP Serverに並列クエリを投げ、要約された回答を生成します。これにより、回答時間が平均5分から30秒へと短縮され、顧客満足度が向上しました。

よくある質問

Q1: MCP Serverを公開する際のセキュリティ対策はどうすればよいですか？

MCP Serverを社内ネットワークやインターネットに公開する場合、認証と認可の仕組みが不可欠です。ローカル開発環境であればstdio（プロセス間通信）を使用するため、物理的なマシンアクセスが制限されていれば比較的安全です。しかし、ネットワーク経由（SSE/HTTP）で公開する場合は、MCPサーバーの前段にAPI GatewayやReverse Proxyを配置し、JWTやOAuth2.0によるトークン検証を行うことを推奨します。また、MCP Server内部の各ツールハンドラーで、渡された引数に基づいた権限チェック（例：ユーザーAがユーザーBのデータを見れないようにする）を必ず実装してください。入力バリデーションもSQLインジェクションやOSコマンドインジェクション対策として重要です。

Q2: 既存のREST APIをMCP Server化するにはどうすればよいですか？

既存のREST APIを捨てる必要はありません。MCP Serverを「アダプター層」として実装し、その内部で既存のREST APIを呼び出す設計が一般的です。例えば、get_order_status というMCPツールを作成し、そのハンドラー関数内で fetch('https://api.internal.com/orders/123') のように既存APIを叩きます。これにより、バックエンドシステムの改修を最小限に抑えつつ、AIエージェントとの統合を迅速に行えます。将来的にマイクロサービス化を進める際も、MCPインターフェースを維持したまま、バックエンドの実装を徐々に置き換えていくことが可能です。

Q3: MCP Serverのパフォーマンスはどの程度ですか？遅延が気になる場合は？

MCP自体のオーバーヘッドは非常に小さく、JSON-RPCのパースにかかる時間は無視できるレベルです。パフォーマンスのボトルネックは、ほぼ間違いなくツールが実行する処理（DBクエリ、外部API呼び出し、ファイルIO）にあります。遅延を気にする場合は、ツールを「並列実行」する設計にすることが重要です。Claudeなどのクライアントは、依存関係のない複数のツール呼び出しを並列で実行する能力を持っています。また、DBクエリの最適化や、キャッシュ戦略（Redisなど）をMCP Server内に導入することで、AIエージェントとの対話におけるレスポンスタイムを大幅に改善できます。

おわりに

Claude CodeとMCP Serverを活用することで、AIエージェントは単なるチャットボットから、開発環境やビジネスシステムに深く統合された「インテリジェントなアシスタント」へと進化します。本記事で解説した実装例は、すべて実務で応用可能な基盤となるものです。特に、Node.jsにおける AsyncLocalStorage を用いたコンテキスト管理や、Prisma/SQLAlchemyの最新機能を活用したデータアクセスは、堅牢で保守性の高いMCP Serverを構築する上で不可欠な要素です。

AI技術の進化は目まぐるしく、開発者が生き残っていくためには、これらのツールを単に使うだけでなく、自社の課題に合わせて拡張・統合する技術力が求められます。弊社Shineosでは、最新のAI技術と既存システムの融合支援を行っております。エージェント活用の第一歩を踏み出したい、あるいは自社システムのAI化をご検討の際は、ぜひお気軽にご相談ください。

参考リンク

[1] Model Context Protocol (MCP) Specification [2] Anthropic Claude Code Documentation [3] Prisma Client Extensions [4] SQLAlchemy 2.0 Documentation [5] AsyncLocalStorage API (Node.js)