TL;DR

個人開発の電気プラン推薦サービス「エネジェント」 (https://enegent.jp) に、MCP (Model Context Protocol) サーバー を同居させて、 Claude Desktop / Cursor 等の外部 AI から自分のサービスを呼べるようにしました
実装方式は FastAPI + FastMCP の同一 Cloud Run コンテナ同居型。 LINE Webhook /api/chat (Web) /mcp の 3 経路が 同じ ToolSpec を参照する single source of truth 設計です
ハマりどころは 3 つで、 (1) FastAPI に mount したサブアプリの lifespan は呼ばれない (2) Cloud Run のホスト名が DNS rebinding 保護で弾かれる (3) def f(**kwargs) だと inputSchema が「kwargs: string」 1 つになる
全部 60 行未満の mcp_server.py で解決できました。個人開発の Free 枠運営でも MCP は普通に乗ります

1. なぜ MCP サーバー化したか

エネジェントは経産省登録の小売電気事業者 800 社超をマスタ DB に取り込み、そのうち 49 社・345 プランを約款ベースで構造化して比較する個人開発サービスです。 Cloud Run + Supabase + Vercel の Free 枠で月 0 円運営しています。

これまで AI 会話 UI は LINE Bot 単独 でした。 Web チャットを足したいというのもあったのですが、同時に「Claude Desktop や Cursor から自分のサービスを呼ばせたい」という欲求がありました。

理由は 2 つあります。

ユーザーが普段使っている AI クライアントから「東京の 30A・月 1 万円ぐらいで一番安いプラン教えて」と聞いたときに、 Web 検索ではなく 約款ベースで構造化済みの自社データ を返したい
個人開発サービスを MCP 化しておくと、将来「AI が代理で比較・推薦する」流れに乗せやすい

MCP は 2025 年に Anthropic が提唱し、 2026 年現在は Claude Desktop / Cursor / Cline / VS Code Copilot Chat 等が対応する事実上の標準プロトコルになっています。「自分のサービスを LLM の道具にする」ことに、そろそろ個人開発でも投資して良いタイミングだと判断しました。

2. 設計の中心: ToolSpec を single source of truth に

LINE Bot / Web チャット / MCP の 3 経路が、 全く同じツールセット を呼び出せる構造にしたかったので、中間表現として ToolSpec という dataclass を 1 つだけ作りました。

# bot/tools.py (抜粋)
from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class ToolParam:
    name: str
    json_type: str  # "string" | "number" | "integer" | "boolean"
    description: str
    required: bool = True
    enum: list[str] | None = None

@dataclass
class ToolSpec:
    name: str
    description: str
    params: list[ToolParam]
    handler: Callable[..., dict]

    def to_json_schema(self) -> dict:
        """MCP / OpenAPI 用 JSON Schema (object) を返す"""
        props, required = {}, []
        for p in self.params:
            d: dict[str, Any] = {"type": p.json_type, "description": p.description}
            if p.enum:
                d["enum"] = p.enum
            props[p.name] = d
            if p.required:
                required.append(p.name)
        return {"type": "object", "properties": props, "required": required}

    def to_gemini_declaration(self) -> dict:
        """Gemini function declaration 形式 (OBJECT/STRING/NUMBER 大文字) に変換"""
        type_map = {"string": "STRING", "number": "NUMBER",
                    "integer": "NUMBER", "boolean": "BOOLEAN"}
        # ...略 (同じ params から大文字型に変換)

ToolSpec から Gemini 用 function declaration にも、 MCP 用 JSON Schema にも変換できる形にしておくのがポイントです。 Gemini は OBJECT/STRING/NUMBER (大文字)、 MCP は object/string/number (小文字 + JSON Schema 形式) と微妙に違うので、 1 ヶ所で吸収しないと両方を保守するのが地獄になります。

公開ツールは 4 つです。

TOOL_SPECS: list[ToolSpec] = [
    ToolSpec(name="recommend_plans", ...),     # エリア・契約状況から最適プラン上位5件
    ToolSpec(name="calc_monthly_bill", ...),   # 指定プラン・月の請求額内訳
    ToolSpec(name="calc_fuel_unit", ...),      # 燃料費調整単価 (中央値+レンジ)
    ToolSpec(name="list_plans", ...),          # エリア・社別のプラン一覧
]

TOOL_BY_NAME: dict[str, ToolSpec] = {t.name: t for t in TOOL_SPECS}

def dispatch_tool(name: str, args: dict[str, Any]) -> dict:
    spec = TOOL_BY_NAME[name]
    return spec.handler(**args)

LINE Bot も Web /api/chat も MCP も、最終的にこの dispatch_tool(name, args) を呼ぶだけ。ツール本体のロジック (_tool_recommend_plans 等) は 完全に共通 です。

3. FastMCP を FastAPI に同居させる

FastMCP は MCP 公式 Python SDK の中の高レベル API です。既存の FastAPI アプリに mount する形で同居させます。

ナイーブに書くとこんな感じになります。

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("enegent")
mcp.add_tool(handler=..., name="recommend_plans", description="...")

app = FastAPI()
app.mount("/mcp", mcp.streamable_http_app())  # ← これだと動かない

これで動けば話は早いのですが、起動して /mcp/mcp を叩くと 500 エラー になります。

RuntimeError: Task group is not initialized. Make sure to use run().

これがハマりどころその 1 です。

ハマり 1: mount したサブアプリの lifespan は呼ばれない

FastMCP.streamable_http_app() は内部で session_manager を持っていて、これを lifespan で起動する必要があります。ところが、 Starlette の app.mount(path, sub_app) で mount した サブアプリの lifespan は呼ばれません (Starlette の仕様)。

解決策: FastAPI 本体の lifespan に MCP の session_manager を組み込む。

# bot/mcp_server.py (抜粋)
from contextlib import AsyncExitStack, asynccontextmanager
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

_streamable_mcp: FastMCP | None = None

def _get_streamable_mcp() -> FastMCP:
    global _streamable_mcp
    if _streamable_mcp is None:
        _streamable_mcp = _build_fastmcp()
        _streamable_mcp.streamable_http_app()  # session_manager の lazy 初期化
    return _streamable_mcp

@asynccontextmanager
async def mcp_lifespan(app: FastAPI):
    """FastAPI の lifespan に組み込む MCP session manager 起動コンテキスト。"""
    streamable_mcp = _get_streamable_mcp()
    async with AsyncExitStack() as stack:
        await stack.enter_async_context(streamable_mcp.session_manager.run())
        yield

def register_mcp(app: FastAPI) -> None:
    streamable_mcp = _get_streamable_mcp()
    streamable_app = streamable_mcp.streamable_http_app()
    streamable_app.add_middleware(BearerAuthMiddleware)
    app.mount("/mcp", streamable_app)

呼び出し側はこうします。

# bot/main.py (抜粋)
from .mcp_server import mcp_lifespan, register_mcp

app = FastAPI(title="エネジェント", lifespan=mcp_lifespan)  # ← lifespan を渡す
register_mcp(app)                                              # ← その後 mount

ポイントは「FastMCP インスタンスを mount 用と lifespan 用で同じものにする」こと。 _streamable_mcp をモジュール変数で保持して _get_streamable_mcp() で取り回すのは、別インスタンスを掴んで session_manager がチグハグになるのを防ぐためです。

ハマり 2: Cloud Run のホスト名が DNS rebinding 保護で弾かれる

ローカルで動かして OK、 Cloud Run にデプロイしたら Invalid Host header で 400 が返る、という古典的な事故が起きました。

FastMCP の TransportSecurityMiddleware はデフォルトで localhost のみ許可 になっています。 Cloud Run のホスト名 (enegent-bot-rw7vp6bhxq-an.a.run.app のような自動生成名) は当然弾かれます。

DNS rebinding 攻撃対策の機能なので無効化するのは怖いのですが、 Cloud Run 側で別途 Bearer 認証を強制するのであれば外して OK です。

# bot/mcp_server.py (抜粋)
from mcp.server.transport_security import TransportSecuritySettings

def _build_fastmcp() -> FastMCP:
    sec = TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
        allowed_hosts=["*"],
        allowed_origins=["*"],
    )
    mcp = FastMCP("enegent", transport_security=sec)
    for spec in TOOL_SPECS:
        handler = _make_handler(spec)
        mcp.add_tool(handler, name=spec.name, description=spec.description)
    return mcp

代わりに、自前の Bearer 認証ミドルウェアを mount 時に噛ませます。

class BearerAuthMiddleware(BaseHTTPMiddleware):
    """Authorization: Bearer <MCP_API_TOKEN> をチェック。
    MCP_API_TOKEN 未設定のとき (開発環境) は auth スキップ。
    """
    async def dispatch(self, request, call_next):
        expected = os.getenv("MCP_API_TOKEN", "")
        if expected:
            auth = request.headers.get("authorization", "")
            if not auth.startswith("Bearer ") or auth[7:] != expected:
                return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

def register_mcp(app: FastAPI) -> None:
    streamable_app = _get_streamable_mcp().streamable_http_app()
    streamable_app.add_middleware(BearerAuthMiddleware)  # ← ここで認証
    app.mount("/mcp", streamable_app)

MCP_API_TOKEN を Cloud Run の環境変数にセットしておけば、これだけで本番で動きます。 Claude Desktop 側の設定では Authorization: Bearer <token> を渡す形になります。

ハマり 3: `**kwargs` だと inputSchema が「kwargs: string」 1 つになる

これが一番気付きにくいやつです。

最初こう書いていました。

def _make_handler(spec):
    def handler(**kwargs):
        result = dispatch_tool(spec.name, kwargs)
        return json.dumps(result, ensure_ascii=False, default=str)
    return handler

mcp.add_tool(handler, name=spec.name, description=spec.description)

これだと、 Claude Desktop から mcp/list_tools で見たときに、各ツールの inputSchema が

{ "type": "object", "properties": { "kwargs": { "type": "string" } } }

になってしまい、 area ampere monthly_kwh 等の個別引数が 一切受け取れません。

原因は FastMCP が inspect.signature(handler) を見て inputSchema を自動生成していること。 **kwargs だと「kwargs という名前の引数 1 つ」として認識されます。

解決策は、 ToolSpec.params から inspect.Signature を 動的に構築 して handler.__signature__ にセットすることです。

# bot/mcp_server.py (抜粋)
def _make_handler(spec):
    import inspect
    type_map = {"string": str, "number": float, "integer": int, "boolean": bool}

    parameters = []
    for p in spec.params:
        annotation = type_map.get(p.json_type, str)
        if p.required:
            param = inspect.Parameter(
                p.name, kind=inspect.Parameter.KEYWORD_ONLY,
                annotation=annotation,
            )
        else:
            param = inspect.Parameter(
                p.name, kind=inspect.Parameter.KEYWORD_ONLY,
                annotation=annotation,
                default=None,  # optional は default=None で表現
            )
        parameters.append(param)
    new_sig = inspect.Signature(parameters=parameters)

    def handler(**kwargs):
        # None で渡された optional 引数は dispatch 前に除去 (handler 側 default を効かせる)
        cleaned = {k: v for k, v in kwargs.items() if v is not None}
        result = dispatch_tool(spec.name, cleaned)
        return json.dumps(result, ensure_ascii=False, default=str)

    handler.__signature__ = new_sig
    handler.__name__ = spec.name
    handler.__doc__ = spec.description
    return handler

KEYWORD_ONLY にしてあるのは、 MCP のツール呼び出しがキーワード引数のみだからです。検証として MCP list_tools を叩くと、期待通りになります。

recommend_plans:
  area (required): エリア識別子 (hokkaido / tohoku / tokyo / ...)
  ampere (optional): 契約アンペア (10A / 15A / 20A / 30A / ...)
  monthly_kwh (optional): 月の使用量 kWh
  monthly_yen (optional): 月の請求額 円
calc_monthly_bill:
  plan_id (required): プラン ID (UUID)
  area (required): エリア識別子
  ampere (required): 契約アンペア
  kwh (required): 月の使用量 kWh
  year (required): 年
  month (required): 月

これで Claude Desktop からも Cursor からも、個別引数で呼べるようになりました。

4. 全体構成: 3 経路が 1 ToolSpec を共有する

設計の全体像を図にするとこうなります。

                              enegent-bot (Cloud Run, asia-northeast1, 1 container)
                              ┌─────────────────────────────────────────────────────────┐
LINE Webhook ─────────────→   │ /webhook         (LINE Messaging API)                   │
                              │                                                          │
Vercel (enegent.jp/chat-mcp)  │                                                          │
  ChatFullPage.tsx ─────────→ │ /api/chat        (Web 用 SSE なし同期版)                │
  fetch + JSON                │                                                          │
                              │   ↓ 3 経路すべて                                         │
                              │ ┌─────────────────────────────────┐                      │
                              │ │ Gemini agent loop               │                      │
                              │ │   tools = bot/tools.TOOL_SPECS  │                      │
                              │ │   .to_gemini_declaration()      │                      │
                              │ └─────────────────────────────────┘                      │
                              │                                                          │
Claude Desktop / Cursor ──→   │ /mcp/mcp         (Streamable HTTP transport)            │
  Bearer MCP_API_TOKEN        │ /mcp-sse/sse     (旧式 SSE transport)                   │
                              │   ↓ register_mcp(app) — bot/mcp_server.py               │
                              │ FastMCP (mcp.server.fastmcp) を FastAPI に mount        │
                              │   ↓                                                      │
                              │ bot/tools.py.dispatch_tool(name, args) 共通ディスパッチ │
                              └─────────────────────────────────────────────────────────┘
                                          ↓
                              Supabase Postgres + pgvector

/mcp/mcp は Streamable HTTP transport (新方式)、 /mcp-sse/sse は SSE transport (旧式互換) を出しています。 Claude Desktop の対応バージョン差異に対する保険として両方公開しています。

mount するときに片方ずつ別の FastMCP インスタンスを使うことに注意が必要です (lifespan の都合)。

def register_mcp(app: FastAPI) -> None:
    streamable_mcp = _get_streamable_mcp()
    sse_mcp = _get_sse_mcp()

    streamable_app = streamable_mcp.streamable_http_app()
    streamable_app.add_middleware(BearerAuthMiddleware)
    app.mount("/mcp", streamable_app)

    sse_app = sse_mcp.sse_app()
    sse_app.add_middleware(BearerAuthMiddleware)
    app.mount("/mcp-sse", sse_app)

5. Claude Desktop からの接続設定

Claude Desktop の claude_desktop_config.json にこう書きます。

{
  "mcpServers": {
    "enegent": {
      "url": "https://your-cloud-run-host/mcp/mcp",
      "headers": {
        "Authorization": "Bearer YOUR_MCP_API_TOKEN"
      }
    }
  }
}

接続後に「東京の 30A で月 1 万円ぐらいの世帯、おすすめプラン上位 5 つ教えて」と話しかけると、 Claude が自動で recommend_plans(area="tokyo", ampere="30A", monthly_yen=10000) を呼んでくれて、約款ベースの構造化済みデータが返ってくる、という体験になります。

Cursor から繋ぐ場合も ~/.cursor/mcp.json に同じ形式で書きます。

6. デプロイと運用の現実

Free 枠で MCP を動かす上での注意点

エネジェントは Cloud Run の Free 枠 で動かしているので、普段は min-instances=0 (リクエストが来たらコールドスタート) です。

これと MCP の相性は微妙で、 Claude Desktop が list_tools を叩く瞬間にコールドスタートが走ると数秒待たされます。とはいえ MCP の用途は対話的な使い方が中心なので、 1 回目のレイテンシは許容範囲です。

レイテンシが気になる場合は、 Web LP 側で <Link onMouseEnter={warmup}> を仕込んでおくと、ユーザーが CTA に hover した瞬間にコンテナが起き、その後の MCP リクエストが速くなる効果も狙えます (これは別記事ネタ)。

`mcp>=1.10.0` と `sse-starlette>=2.1.0`

requirements-deploy.txt にはこの 2 つを足しておく必要があります。 Streamable HTTP は mcp>=1.10.0、 SSE 互換用に sse-starlette も入れておきます。

mcp>=1.10.0
sse-starlette>=2.1.0
fastapi>=0.111

lifespan が失敗しても本体は落としたくない

bot/main.py 側はこういうフェイルセーフを書いてあります。 MCP の初期化が何かの拍子に失敗しても、 LINE Webhook と /api/chat は止めたくないので。

try:
    from .mcp_server import mcp_lifespan, register_mcp
    app = FastAPI(title="エネジェント", lifespan=mcp_lifespan)
    _MCP_ENABLED = True
except Exception as e:
    logger.warning("MCP lifespan setup failed, MCP disabled: %s", e)
    app = FastAPI(title="エネジェント")
    _MCP_ENABLED = False

if _MCP_ENABLED:
    try:
        register_mcp(app)
    except Exception as e:
        logger.warning("MCP server mount skipped: %s", e)

個人開発で全部 1 コンテナに同居させると、こういうフォールバック設計を書いておかないと「MCP の依存ライブラリが壊れた瞬間に LINE Bot 全停止」みたいな事故が起きます。

7. やってみての教訓

MCP 化のコストはほぼゼロ。既存の FastAPI バックエンドに 178 行の mcp_server.py を 1 ファイル足すだけで終わりました
ToolSpec を中間表現にしておくと、 LLM 経由とプロトコル経由の二重定義から解放される。ツール仕様の不整合を構造的に防げます
個人開発でも、自分のサービスを「LLM が呼ぶ道具」にしておくと将来の選択肢が広がる。比較系・推薦系・データ系のサービスは特に MCP との相性が良いと感じました
ハマりどころは全部 lifespan / DNS rebinding / inspect.signature の 3 つ。ここだけ押さえれば後は素直に動きます

エネジェント (https://enegent.jp) は完全に趣味の個人プロジェクトで、月 0 円で運営しています。「自分のサービスを Claude や Cursor から呼ばせてみる」のは、やってみると思っていたより簡単で、思っていたより楽しいです。 MCP 対応の波に乗ってみたい個人開発者の方の参考になれば。

過去記事も合わせてどうぞ:

4-layer のチャットエージェントを「Single Agent + Native Tool Calling」に作り直したら、ハルシネーションも文脈分断も消えた話 — チャット V1 → V2 の話
経産省登録 808 社の電力料金表を LLM で自動構造化するパイプラインを作った — データ基盤の話
Gemini function calling で LINE Bot を「AI エージェント」にする実装パターン — LINE Bot のスケルトン