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 のスケルトン