MCPサーバー自作入門【2026】広告データ・運用ナレッジをClaude Codeにつなぐ（TypeScript/Python実装）

mkdir ad-insights-mcp && cd ad-insights-mcp
npm init -y
npm pkg set type=module
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
mkdir src

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./build",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true
  },
  "include": ["src/**/*"]
}

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

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

server.registerTool(
  "add",
  {
    title: "Add",
    description: "Add two numbers",
    inputSchema: { a: z.number(), b: z.number() },
  },
  async ({ a, b }) => ({ content: [{ type: "text", text: String(a + b) }] })
);

const transport = new StdioServerTransport();
await server.connect(transport);
console.error("demo MCP server running on stdio"); // ログは stderr へ

import { readFile, readdir } from "node:fs/promises";
import path from "node:path";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

// データの置き場所（取得スクリプトが書き出したCSV・運用ナレッジのMarkdown）
const DATA_DIR = process.env.AD_DATA_DIR ?? "./data";
const KNOWLEDGE_DIR = process.env.AD_KNOWLEDGE_DIR ?? "./knowledge";
const PAGE_SIZE = 10;

type Row = {
  date: string; platform: string; campaign_name: string;
  impressions: number; clicks: number; cost: number; conversions: number;
};

// 取得済みCSV（date,platform,campaign_id,campaign_name,impressions,clicks,cost,conversions）を読む
async function loadRows(platform: "google_ads" | "meta"): Promise<Row[]> {
  const file = path.join(DATA_DIR, `${platform}_daily.csv`);
  const text = await readFile(file, "utf-8"); // 無ければ例外 → 呼び出し側で isError に
  const [header, ...lines] = text.trim().split(/\r?\n/);
  const cols = header.split(",");
  return lines.map((line) => {
    const v = line.split(","); // 例示のため簡易パース。本番はCSVライブラリかDBを使う
    const get = (k: string) => v[cols.indexOf(k)];
    return {
      date: get("date"), platform, campaign_name: get("campaign_name"),
      impressions: Number(get("impressions")), clicks: Number(get("clicks")),
      cost: Number(get("cost")), conversions: Number(get("conversions")),
    };
  });
}

const errorResult = (message: string) => ({
  content: [{ type: "text" as const, text: message }],
  isError: true, // ツールの失敗はプロトコルエラーではなく「結果」として返す
});

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

// ツール1：キャンペーン実績の集計（API1本=ツール1個ではなく「知りたいこと」単位）
server.registerTool(
  "get_campaign_performance",
  {
    title: "キャンペーン実績の集計",
    description:
      "Google広告またはMeta広告のキャンペーン別実績（費用・CV・CPA・CTR・CVR）を、指定期間で集計して返す。" +
      "『どのキャンペーンが悪化したか』『費用の大きい順』などを知りたいときに使う。" +
      "日別の生データは返さない。結果が多い場合は next_cursor を使って続きを取得する。" +
      "読み取り専用で、広告アカウントには一切変更を加えない。",
    inputSchema: {
      platform: z.enum(["google_ads", "meta"]).describe("媒体"),
      days: z.number().int().min(1).max(90).default(7).describe("昨日から遡る日数"),
      sort_by: z.enum(["cost", "cpa", "conversions"]).default("cost").describe("並び順（降順）"),
      cursor: z.number().int().min(0).default(0).describe("ページング位置。初回は0"),
    },
    annotations: { readOnlyHint: true },
  },
  async ({ platform, days, sort_by, cursor }) => {
    let rows: Row[];
    try {
      rows = await loadRows(platform);
    } catch {
      return errorResult(
        `${platform} のデータが見つかりません。先に取得スクリプト（fetch_${platform}.py）を実行してください。`
      );
    }
    const dates = [...new Set(rows.map((r) => r.date))].sort();
    const target = new Set(dates.slice(-days));

    const byCampaign = new Map<string, { cost: number; cv: number; imp: number; clk: number }>();
    for (const r of rows.filter((r) => target.has(r.date))) {
      const a = byCampaign.get(r.campaign_name) ?? { cost: 0, cv: 0, imp: 0, clk: 0 };
      a.cost += r.cost; a.cv += r.conversions; a.imp += r.impressions; a.clk += r.clicks;
      byCampaign.set(r.campaign_name, a);
    }
    const summary = [...byCampaign.entries()].map(([name, a]) => ({
      campaign: name,
      cost: Math.round(a.cost),
      conversions: Math.round(a.cv * 10) / 10,
      cpa: a.cv > 0 ? Math.round(a.cost / a.cv) : null, // CV0件は null（0円と誤読させない）
      ctr: a.imp > 0 ? +(a.clk / a.imp * 100).toFixed(2) : null,
      cvr: a.clk > 0 ? +(a.cv / a.clk * 100).toFixed(2) : null,
    }));
    summary.sort((x, y) => (y[sort_by] ?? -1) - (x[sort_by] ?? -1));

    const page = summary.slice(cursor, cursor + PAGE_SIZE);
    const next = cursor + PAGE_SIZE < summary.length ? cursor + PAGE_SIZE : null;
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          platform, period: `${[...target][0]}〜${[...target].at(-1)}`,
          total_campaigns: summary.length, campaigns: page, next_cursor: next,
          note: "cost は円、ctr/cvr は%。cpa=cost/conversions",
        }),
      }],
    };
  }
);

// ツール2：運用ナレッジ（社内の運用ルール・過去の施策メモ）の検索
server.registerTool(
  "search_ops_knowledge",
  {
    title: "運用ナレッジ検索",
    description:
      "社内の広告運用ナレッジ（アカウントごとの運用ルール、CV定義、過去の施策と結果のメモ）をキーワード検索し、" +
      "該当箇所の抜粋を最大5件返す。数値の変動の原因仮説を立てる前や、改善提案をする前に使う。",
    inputSchema: {
      query: z.string().min(1).describe("検索語（例: 'P-MAX 除外', 'CV定義 Meta'）"),
    },
    annotations: { readOnlyHint: true },
  },
  async ({ query }) => {
    const words = query.toLowerCase().split(/\s+/);
    const hits: string[] = [];
    for (const f of await readdir(KNOWLEDGE_DIR)) {
      if (!f.endsWith(".md")) continue;
      const text = await readFile(path.join(KNOWLEDGE_DIR, f), "utf-8");
      for (const para of text.split(/\n{2,}/)) {
        if (words.every((w) => para.toLowerCase().includes(w))) {
          hits.push(`[${f}] ${para.slice(0, 400)}`); // 抜粋は短く
        }
      }
    }
    return {
      content: [{
        type: "text",
        text: hits.length ? hits.slice(0, 5).join("\n---\n") : "該当するナレッジはありません。",
      }],
    };
  }
);

// リソース：指標の定義（Claudeや人が「読む」ためのデータ）
server.registerResource(
  "metric-definitions",
  "docs://ad-insights/metrics",
  { title: "指標の定義", description: "CPA・CVR等の計算式と注意点", mimeType: "text/markdown" },
  async (uri) => ({
    contents: [{
      uri: uri.href,
      text: "- CPA = cost / conversions（CV0件は算出しない）\n- CTR = clicks / impressions\n- CVR = conversions / clicks\n- Google広告の費用は cost_micros / 1,000,000",
    }],
  })
);

await server.connect(new StdioServerTransport());
console.error("ad-insights MCP server running on stdio");

# クライアントA 運用メモ（knowledge/client_a.md）

## CV定義
Google広告は「購入完了」のみをメインCVにしている。Meta の CV定義は purchase。

## 過去の施策
2026年8月、P-MAX に除外キーワード（指名系）を追加。指名検索キャンペーンとの食い合いを抑える目的。

"""Google広告の実績をMCPツールとして公開する（読み取り専用）"""
import os
import sys
from functools import cache
from typing import Annotated, Literal

from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException
from mcp.server.mcpserver import MCPServer  # v1系: from mcp.server.fastmcp import FastMCP
from mcp.server.mcpserver.exceptions import ToolError
from pydantic import Field

mcp = MCPServer("google-ads-readonly")


@cache
def ads_client() -> GoogleAdsClient:
    """初回呼び出し時に1回だけ認証（起動時に認証すると、失敗でサーバーごと落ちる）"""
    return GoogleAdsClient.load_from_storage(os.environ["GOOGLE_ADS_CONFIG"])

# 触ってよいアカウントを許可リストで固定（モデルに任意のIDを指定させない）
ALLOWED_CUSTOMERS = set(os.environ["GOOGLE_ADS_ALLOWED_CUSTOMERS"].split(","))


@mcp.tool()
def get_campaign_performance(
    customer_id: Annotated[str, Field(description="広告アカウントID（10桁）。許可されたアカウントのみ")],
    period: Annotated[
        Literal["LAST_7_DAYS", "LAST_14_DAYS", "LAST_30_DAYS"], Field(description="集計期間")
    ] = "LAST_7_DAYS",
    limit: Annotated[int, Field(ge=1, le=50, description="返すキャンペーン数の上限")] = 20,
) -> str:
    """Google広告のキャンペーン別実績（費用・CV・CPA）を費用の大きい順に返す。
    『今週どのキャンペーンにお金が使われ、CPAはいくらか』を知りたいときに使う。読み取り専用。"""
    customer_id = customer_id.replace("-", "")
    if customer_id not in ALLOWED_CUSTOMERS:
        # ToolError はメッセージ付きの isError: true としてモデルに返る
        raise ToolError(f"アカウント {customer_id} は許可されていません")

    query = f"""
        SELECT campaign.name, metrics.cost_micros, metrics.conversions, metrics.clicks
        FROM campaign
        WHERE segments.date DURING {period} AND campaign.status = 'ENABLED'
        ORDER BY metrics.cost_micros DESC
        LIMIT {limit}
    """
    try:
        service = ads_client().get_service("GoogleAdsService")
        lines = []
        for batch in service.search_stream(customer_id=customer_id, query=query):
            for row in batch.results:
                cost = row.metrics.cost_micros / 1_000_000
                cv = row.metrics.conversions
                cpa = f"{cost / cv:,.0f}円" if cv > 0 else "CVなし"
                lines.append(f"{row.campaign.name}: 費用{cost:,.0f}円 / CV{cv:.1f} / CPA {cpa}")
    except GoogleAdsException as ex:
        messages = "; ".join(e.message for e in ex.failure.errors)
        raise ToolError(f"Google Ads API エラー: {messages}") from ex

    return "\n".join(lines) if lines else "該当期間に配信中のキャンペーンはありません。"


if __name__ == "__main__":
    print("google-ads-readonly MCP server starting", file=sys.stderr)  # stdout は使わない
    mcp.run()  # 既定は stdio

python3 -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]" google-ads

# 単体テスト：Inspector から起動
npx @modelcontextprotocol/inspector python server.py

# Claude Code に登録（秘密情報は -e で渡し、コードやGitに書かない）
claude mcp add google-ads-readonly \
  -e GOOGLE_ADS_CONFIG=$HOME/.config/ad-analytics/google-ads.yaml \
  -e GOOGLE_ADS_ALLOWED_CUSTOMERS=1234567890 \
  -- /path/to/.venv/bin/python /path/to/server.py

npx tsc                                   # src/ → build/ にコンパイル
npx @modelcontextprotocol/inspector node build/index.js
# ブラウザで Inspector が開く → Connect → Tools タブで一覧表示・実行

# 自分だけで使う（local スコープ）
claude mcp add ad-insights -- node /path/to/ad-insights-mcp/build/index.js

# 環境変数を渡す
claude mcp add ad-insights -e AD_DATA_DIR=/path/to/data -e AD_KNOWLEDGE_DIR=/path/to/knowledge \
  -- node /path/to/ad-insights-mcp/build/index.js

# チームで共有する（プロジェクト直下の .mcp.json に保存 → Git で共有）
claude mcp add ad-insights --scope project -- node ./mcp/ad-insights/build/index.js

# 確認
claude mcp list
# Claude Code の中では /mcp で接続状況とツール一覧を確認できる

{
  "mcpServers": {
    "ad-insights": {
      "command": "node",
      "args": ["./mcp/ad-insights/build/index.js"],
      "env": {
        "AD_DATA_DIR": "./data",
        "AD_KNOWLEDGE_DIR": "./knowledge"
      }
    },
    "google-ads-readonly": {
      "command": "python",
      "args": ["./mcp/google_ads/server.py"],
      "env": {
        "GOOGLE_ADS_CONFIG": "${GOOGLE_ADS_CONFIG}",
        "GOOGLE_ADS_ALLOWED_CUSTOMERS": "${GOOGLE_ADS_ALLOWED_CUSTOMERS}"
      }
    }
  }
}

{
  "mcpServers": {
    "ad-insights": {
      "command": "node",
      "args": ["/Users/you/ad-insights-mcp/build/index.js"],
      "env": { "AD_DATA_DIR": "/Users/you/ad-analytics/data" }
    }
  }
}

import express from "express";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { z } from "zod";

function buildServer(): McpServer {
  const server = new McpServer({ name: "ad-insights-remote", version: "1.0.0" });
  server.registerTool(
    "ping",
    { title: "Ping", description: "疎通確認用", inputSchema: { message: z.string() } },
    async ({ message }) => ({ content: [{ type: "text", text: `pong: ${message}` }] })
  );
  return server;
}

const app = express();
app.use(express.json());

// ステートレス構成：リクエストごとにサーバーとトランスポートを作る
app.post("/mcp", async (req, res) => {
  // 本番ではここより前段で OAuth のアクセストークンを検証する（後述）
  const server = buildServer();
  const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined });
  res.on("close", () => {
    transport.close();
    server.close();
  });
  await server.connect(transport);
  await transport.handleRequest(req, res, req.body);
});

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

# リモート（Streamable HTTP）の MCP サーバーを登録
claude mcp add --transport http ad-insights-remote https://mcp.example.com/mcp

# OAuth 対応サーバーなら、Claude Code 内の /mcp から認証フローを開始できる

import anthropic

client = anthropic.Anthropic()

# Claude API 側から、公開済みのリモートMCPサーバーを直接使う（beta）
response = client.beta.messages.create(
    model="claude-opus-5",
    max_tokens=16000,
    betas=["mcp-client-2025-11-20"],
    mcp_servers=[{
        "type": "url",
        "url": "https://mcp.example.com/mcp",
        "name": "ad-insights",
        # "authorization_token": "...",  # OAuth で取得したアクセストークン
    }],
    tools=[{"type": "mcp_toolset", "mcp_server_name": "ad-insights"}],
    messages=[{"role": "user", "content": "先週のGoogle広告でCPAが高い順に5キャンペーン教えて"}],
)
for block in response.content:
    if block.type == "text":
        print(block.text)