MCPサーバー構築実践 — 自分のツールをあらゆるAIエージェントにつなぐ方法

+----------------------------------------------------------+
|                      Host (ホスト)                         |
|   Claude Desktop、IDE、カスタムエージェントアプリなど           |
|                                                          |
|  +------------------+        +------------------+        |
|  |  MCP Client #1   |        |  MCP Client #2   |        |
|  +--------+---------+        +--------+---------+        |
+-----------|--------------------------|-------------------+
            | 1:1 接続                  | 1:1 接続
            v                          v
   +------------------+      +------------------+
   |  MCP Server A    |      |  MCP Server B    |
   |  (社内Wiki検索)    |      |  (DB照会ツール)    |
   +------------------+      +------------------+

[stdio 方式]
Host ──(子プロセスとして起動)──> Server
       stdin/stdout で JSON-RPC メッセージを交換

[streamable HTTP 方式]
Host ──HTTP POST /mcp──> Server (単一エンドポイント)
     <──JSON レスポンスまたは SSE ストリーム──

mkdir wiki-mcp-server && cd wiki-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init --target es2022 --module nodenext --moduleResolution nodenext --outDir dist

// src/index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const WIKI_BASE_URL = process.env.WIKI_BASE_URL ?? "https://wiki.internal.example.com";
const WIKI_API_TOKEN = process.env.WIKI_API_TOKEN;

if (!WIKI_API_TOKEN) {
  // stdioサーバーでconsole.logはプロトコルを汚染するため必ずstderrを使う
  console.error("WIKI_API_TOKEN 環境変数が必要です。");
  process.exit(1);
}

const server = new McpServer({
  name: "company-wiki",
  version: "1.0.0",
});

async function wikiFetch(path: string): Promise<unknown> {
  const res = await fetch(`${WIKI_BASE_URL}${path}`, {
    headers: { Authorization: `Bearer ${WIKI_API_TOKEN}` },
  });
  if (!res.ok) {
    throw new Error(`Wiki APIエラー: HTTP ${res.status} ${res.statusText}`);
  }
  return res.json();
}

// ツール1: Wiki検索
server.registerTool(
  "search_wiki",
  {
    title: "社内Wiki検索",
    description:
      "社内Wikiでドキュメントを検索します。タイトルと本文を対象にキーワード検索を行い、" +
      "ページID、タイトル、要約、最終更新日を返します。" +
      "ページ全文が必要な場合は read_wiki_page ツールを続けて使用してください。",
    inputSchema: {
      query: z.string().min(1).describe("検索キーワード。例: 'デプロイ手順', 'VPN設定'"),
      limit: z.number().int().min(1).max(20).default(5)
        .describe("最大結果数 (1-20、デフォルト5)"),
    },
  },
  async ({ query, limit }) => {
    const data = (await wikiFetch(
      `/api/v1/search?q=${encodeURIComponent(query)}&limit=${limit}`
    )) as { results: Array<{ id: string; title: string; excerpt: string; updatedAt: string }> };

    if (data.results.length === 0) {
      return {
        content: [{
          type: "text",
          text: `'${query}' に対する検索結果がありません。より短い中核キーワードで再試行してください。`,
        }],
      };
    }

    const lines = data.results.map(
      (r) => `- [${r.id}] ${r.title} (更新: ${r.updatedAt})\n  ${r.excerpt}`
    );
    return {
      content: [{ type: "text", text: lines.join("\n") }],
    };
  }
);

// ツール2: ページ本文の読み取り
server.registerTool(
  "read_wiki_page",
  {
    title: "Wikiページ読み取り",
    description:
      "ページIDを指定して社内Wikiページの全文をマークダウンで取得します。" +
      "ページIDは search_wiki の結果の角括弧内の値です。",
    inputSchema: {
      pageId: z.string().regex(/^[A-Za-z0-9-]+$/).describe("ページID。例: 'deploy-guide-2026'"),
    },
  },
  async ({ pageId }) => {
    const page = (await wikiFetch(`/api/v1/pages/${pageId}`)) as {
      title: string; bodyMarkdown: string;
    };
    return {
      content: [{
        type: "text",
        text: `# ${page.title}\n\n${page.bodyMarkdown}`,
      }],
    };
  }
);

// リソース: Wikiカテゴリ一覧 (アプリがコンテキストとして添付可能)
server.registerResource(
  "wiki-categories",
  "wiki://categories",
  {
    title: "Wikiカテゴリ一覧",
    description: "社内Wikiの最上位カテゴリ一覧",
    mimeType: "text/plain",
  },
  async (uri) => {
    const data = (await wikiFetch("/api/v1/categories")) as { names: string[] };
    return {
      contents: [{ uri: uri.href, text: data.names.join("\n") }],
    };
  }
);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("company-wiki MCPサーバーがstdioで実行中です。");
}

main().catch((err) => {
  console.error("致命的なエラー:", err);
  process.exit(1);
});

npx tsc
claude mcp add company-wiki \
  --env WIKI_API_TOKEN=your-token-here \
  -- node /absolute/path/to/wiki-mcp-server/dist/index.js

{
  "mcpServers": {
    "company-wiki": {
      "command": "node",
      "args": ["/absolute/path/to/wiki-mcp-server/dist/index.js"],
      "env": {
        "WIKI_API_TOKEN": "your-token-here",
        "WIKI_BASE_URL": "https://wiki.internal.example.com"
      }
    }
  }
}

# server.py
import os
import sys
import httpx
from mcp.server.fastmcp import FastMCP

WIKI_BASE_URL = os.environ.get("WIKI_BASE_URL", "https://wiki.internal.example.com")
WIKI_API_TOKEN = os.environ.get("WIKI_API_TOKEN")

if not WIKI_API_TOKEN:
    print("WIKI_API_TOKEN 環境変数が必要です。", file=sys.stderr)
    sys.exit(1)

mcp = FastMCP("company-wiki")


async def wiki_get(path: str) -> dict:
    async with httpx.AsyncClient(
        base_url=WIKI_BASE_URL,
        headers={"Authorization": f"Bearer {WIKI_API_TOKEN}"},
        timeout=10.0,
    ) as client:
        resp = await client.get(path)
        resp.raise_for_status()
        return resp.json()


@mcp.tool()
async def search_wiki(query: str, limit: int = 5) -> str:
    """社内Wikiでドキュメントを検索します。

    タイトルと本文を対象にキーワード検索を行い、ページID、タイトル、要約を返します。
    全文が必要な場合は read_wiki_page ツールを続けて使用してください。

    Args:
        query: 検索キーワード。例: 'デプロイ手順', 'VPN設定'
        limit: 最大結果数 (1-20、デフォルト5)
    """
    data = await wiki_get(f"/api/v1/search?q={query}&limit={limit}")
    results = data.get("results", [])
    if not results:
        return f"'{query}' に対する検索結果がありません。より短い中核キーワードで再試行してください。"
    lines = [
        f"- [{r['id']}] {r['title']} (更新: {r['updatedAt']})\n  {r['excerpt']}"
        for r in results
    ]
    return "\n".join(lines)


@mcp.tool()
async def read_wiki_page(page_id: str) -> str:
    """ページIDを指定してWikiページの全文をマークダウンで取得します。

    Args:
        page_id: ページID。search_wiki の結果の角括弧内の値。例: 'deploy-guide-2026'
    """
    page = await wiki_get(f"/api/v1/pages/{page_id}")
    return f"# {page['title']}\n\n{page['bodyMarkdown']}"


if __name__ == "__main__":
    mcp.run(transport="stdio")

uv init wiki-mcp && cd wiki-mcp
uv add "mcp[cli]" httpx
uv run mcp dev server.py        # Inspectorと一緒に開発モードで実行
uv run mcp install server.py    # Claude Desktopに登録

if __name__ == "__main__":
    mcp.run(transport="streamable-http")  # デフォルトで /mcp エンドポイントを公開

+--------+          +------------------+          +--------------------+
| Client | --1.---> | MCP Server       |          | Authorization      |
|        | <--401-- | (Resource Server)|          | Server (IdP)       |
|        |          +------------------+          +--------------------+
|        | --2. メタデータ取得 (RFC 9728)---------------------^
|        | --3. 動的クライアント登録 + PKCE認可コードフロー------>|
|        | <-4. access token--------------------------------+
|        | --5. Authorization: Bearer トークンでMCPリクエスト-->
+--------+

# TypeScriptサーバー
npx @modelcontextprotocol/inspector node dist/index.js

# Pythonサーバー (mcp devがInspectorも一緒に起動)
uv run mcp dev server.py

// tests/search.test.ts — ビジネスロジックを分離して単体テスト
import { describe, it, expect, vi } from "vitest";
import { formatSearchResults } from "../src/format.js";

describe("formatSearchResults", () => {
  it("空の結果に再試行のヒントを含む", () => {
    const text = formatSearchResults("vpn", []);
    expect(text).toContain("検索結果がありません");
    expect(text).toContain("再試行");
  });

  it("結果をIDとタイトルの一覧に整形する", () => {
    const text = formatSearchResults("deploy", [
      { id: "deploy-guide", title: "デプロイガイド", excerpt: "...", updatedAt: "2026-06-01" },
    ]);
    expect(text).toContain("[deploy-guide]");
  });
});