設計
業務用 AI チャットボットに「不動産情報や地理空間データを自然な会話の中で参照させたい」という要件を実現するため、国土交通省が α 版で公開している地理空間 MCP Server (chirikuuka/mlit-geospatial-mcp) を組み込む案件を検討した。
初期は「公式の MCP を採用すれば AI が裏でいい感じに使ってくれるはず」と楽観視していたが、実際に手を動かしてみると MCP Server は Claude Desktop 想定で作られた stdio 専用の subprocess であり、サーバーサイドの Web サービスから利用するには複数の adaptation が必要だった。本記事はその実装パターンと判断ポイントを整理したものである。
MCP Server を ECS Fargate 上の Web サービスから安定して利用するアーキテクチャ
Claude Desktop 専用 (stdio) の MCP Server を mcp-proxy 経由で HTTP 化する方法
Function Calling ループと MCP セッション管理の実装パターン
Feature Flag を 3 変数連動させて段階的にロールアウトする設計
公式 MCP を業務用途に組み込む際に踏みやすい落とし穴と回避策
バックエンドエンジニアで MCP Server を業務サービスに組み込みたい人
既存の FastAPI / OpenAI SDK ベースの AI チャットボットに MCP を追加したい人
ECS Fargate 上でサイドカーパターンを採用するか迷っている人
「公式 MCP を採用すれば終わり」だと思っていた人（自分）
バックエンド: Python 3.13 + FastAPI、Azure OpenAI SDK で Function Calling を既に利用中
インフラ: AWS ECS Fargate + ecspresso、Terraform で SSM/IAM/ECR を管理
LLM: Azure OpenAI gpt-4o（128k context）
MCP SDK: mcp Python SDK v1.17 以上、Streamable HTTP transport
対象 MCP: chirikuuka/mlit-geospatial-mcp (α版、Python 3.10+, stdio 配布)
まず誤解しがちなポイントだが、MCP Server は AI ではない。LLM が呼び出せるツールの集合 (= 棚) を標準化されたプロトコルで提供する箱にすぎない。「自然な質問を投げたら勝手にいい感じに応える」のは LLM 側の責任で、MCP Server はその LLM が tools/list でツール一覧を取得し、tools/call でツールを呼び出すための薄い層を提供する。
つまり、Web サービスから MCP を使うには「ユーザーの自然文を理解して、どの MCP ツールをどう呼ぶか LLM に決めさせ、結果を整形してまた LLM に戻す」というオーケストレーションを自前で書かなければいけない。
世の中で流通している MCP Server の多くは Claude Desktop / IDE プラグイン (Cursor 等) を想定して作られており、以下の暗黙の仮定を持っている:
stdio トランスポート専用: ホスト (Claude Desktop) が subprocess として起動する想定
同時接続は 1 つだけ: 1 ユーザーの 1 セッションを前提にした単一 client
対話的な確認フロー: 「ファイル保存しますか？」のようにユーザーに聞き返す挙動
ローカルファイルシステム前提: 取得結果を output_dir に保存する機能
業務用 Web サービスでこれらをそのまま使うと、ユーザーが応えられない確認ループに陥ったり、コンテナ内に意味のないファイルが書き出されたり、複数ユーザーの同時アクセスで session が混線する。
MCP Server を ECS タスク内のサイドカーコンテナとして同居させ、メインの FastAPI アプリから localhost で HTTP 接続させる構成を採用した。
flowchart TD
subgraph ECS["ECS Task (1つの論理単位)"]
subgraph API["① api コンテナ (FastAPI)"]
direction TB
MS["MessageService<br/>(chitchat / KB / 地理空間 ルーティング)"]
RS["ReinfolibService"]
CS["MCP ClientSession<br/>(lifespan で1本維持)"]
MS --> RS --> CS
end
subgraph SIDECAR["② reinfolib-mcp コンテナ"]
direction TB
PROXY["mcp-proxy<br/>(stdio↔HTTP 変換)"]
SERVER["chirikuuka/mlit-geospatial-mcp<br/>(Python)"]
PROXY --> SERVER
end
CS -- "localhost:3001/mcp (HTTP)" --> PROXY
end
SERVER -- "HTTPS" --> EXT["外部 API<br/>(国交省 reinfolib)"]
この構成のメリット:
プロセス分離: MCP Server がクラッシュしても api は無傷で、ECS が片方だけ再起動
localhost 通信: 認証不要、低レイテンシ、ネットワーク露出なし
既存運用に乗る: ECS タスク定義に containerDefinitions を 1 つ追加するだけで標準的な ECS 運用に組み込める
公式 MCP Server は stdio 専用なので、そのままだとサイドカーコンテナで HTTP 公開できない。mcp-proxy という OSS が stdio ↔ HTTP / SSE の変換を担ってくれるので、これを Dockerfile で同梱する。
FROM python:3.10-slim
RUN apt-get update && apt-get install -y --no-install-recommends \
git build-essential libgeos-dev libproj-dev \
&& rm -rf /var/lib/apt/lists/*
# mcp-proxy は MCP 本体とは独立した依存なので先に入れてレイヤキャッシュを効かせる
RUN pip install --no-cache-dir mcp-proxy
# 公式 MCP Server を固定 SHA で取得（α版のため予告なき変更に防御）
ARG MCP_COMMIT_SHA=eaf3592c9d1e3c10b294fd4218a32a8ab806ed2c
WORKDIR /opt/mcp
RUN git clone https://github.com/chirikuuka/mlit-geospatial-mcp.git . \
&& git checkout "${MCP_COMMIT_SHA}" \
&& pip install --no-cache-dir -r requirements.txt
WORKDIR /opt/mcp/src
EXPOSE 3001
# stdio で動く本体を Streamable HTTP として公開する。
# --pass-environment でコンテナ env を subprocess (chirikuuka 本体) に継承させる。
CMD ["mcp-proxy", "--transport", "streamablehttp", \
"--port", "3001", "--host", "0.0.0.0", \
"--pass-environment", \
"--", "python", "server.py"]
ここで地味に重要なのが --pass-environment で、これがないと API キー等の環境変数が subprocess に伝わらず 401 を浴びる。
ローカルの docker-compose と本番 ECS で、同じ Dockerfile を別の起動方法で扱う。
ローカル (docker-compose.yml):
services:
app:
environment:
# Docker network ではサービス名で名前解決するので reinfolib-mcp:3001 で届く
REINFOLIB_MCP_URL: http://reinfolib-mcp:3001/mcp
depends_on:
reinfolib-mcp:
condition: service_started
required: false # profile mcp 未指定時に sidecar 不在でも起動可
reinfolib-mcp:
profiles: ["mcp"] # COMPOSE_PROFILES=mcp が指定された時だけ起動
build: ../reinfolib-mcp-sidecar
environment:
LIBRARY_API_KEY: ${REINFOLIB_API_KEY}
ports:
- "3001:3001"
対応する ECS task-definition (task-definition.json):
{
"containerDefinitions": [
{
"name": "api",
"environment": [{ "name": "ENABLE_REINFOLIB_MCP", "value": "true" }],
"dependsOn": [{ "containerName": "reinfolib-mcp", "condition": "START" }]
},
{
"name": "reinfolib-mcp",
"essential": false,
"image": "...:reinfolib-mcp-sidecar",
"secrets": [{ "name": "LIBRARY_API_KEY", "valueFrom": "/.../REINFOLIB_API_KEY" }],
"portMappings": [{ "containerPort": 3001 }]
}
]
}
Fargate の awsvpc ネットワークでは同一タスク内のコンテナが ENI を共有するので、api からは http://localhost:3001/mcp でサイドカーに届く。docker-compose は別ネットワークなのでサービス名 (reinfolib-mcp) で解決する。env の default は ECS 想定値 (localhost) にしておき、ローカル側だけ docker-compose で上書きしている。
ここからが本題。Claude Desktop 専用の MCP Server を「業務 Web サービスでも安定して動く形」にラップするために必要だった処理を 5 つに分けて整理する。
MCP Server を入れても、自然文 → ツール選択 → 結果整形 → 自然文応答のオーケストレーションは LLM 側 (= 我々のコード) の責任。OpenAI SDK の tools= パラメータに MCP から取得したツール一覧を流し込み、tool_choice="auto" で LLM に判断させ、tool_calls が返ってきたら MCP に call_tool して結果を messages に積み直す、という古典的な loop を回す。
async def _run_tool_iterations(self, messages):
tool_schemas = cast(list[ChatCompletionToolParam], self._tool_schemas)
for _ in range(_MAX_TOOL_ITERATIONS):
response = await self._openai.chat.completions.create(
model=_MODEL, temperature=0,
messages=messages, tools=tool_schemas, tool_choice="auto",
)
msg = response.choices[0].message
if not msg.tool_calls:
return # ツール呼び出し終わり、最終応答へ
function_calls = [tc for tc in msg.tool_calls if tc.type == "function"]
messages.append(self._build_assistant_message(msg, function_calls))
for tc in function_calls:
await self._execute_tool_call(tc, messages)
ループ上限 (_MAX_TOOL_ITERATIONS = 8) は target_apis を分割呼び出しする質問への耐性として持たせている。
ここが意外と引っかかったポイント。MCP Streamable HTTP の ClientSession は同一 session で複数の call_tool を同時に走らせると JSON-RPC ID が衝突して壊れる。
FastAPI は asyncio で多数のリクエストを並行処理するので、複数ユーザーが同時に質問すると同一 session を奪い合う。シングルトンで MCP セッションを共有しつつ、call_tool の入口に asyncio.Lock を置いて直列化した。
class ReinfolibService:
def __init__(self) -> None:
self._session: ClientSession | None = None
# MCP ClientSession は同時 call_tool 非対応。
# 複数の FastAPI リクエストが同一 session を奪い合うのを防ぐ
self._call_lock = asyncio.Lock()
async def _invoke_tool(self, name, arguments):
async with self._call_lock:
result = await self._session.call_tool(name, arguments)
return _serialize_mcp_result(result.content)
本番で同時アクセス数が増えてレイテンシが問題になったら、session pool 化や別タスクへの sidecar 切り出しで水平スケールを検討する余地がある。stg レベルなら Lock で十分だった。
公式 MCP のツールには save_file (取得結果をローカルに GeoJSON で保存するか) と output_dir (保存先パス) というパラメータがある。これは Claude Desktop で「ユーザーの PC のダウンロードフォルダに保存」する機能だが、Web チャットでは:
そもそもサーバー側にファイル保存しても意味がない
LLM が save_file=null を渡すと MCP が「保存しますか？」と聞き返してきて、Web では応えるユーザー操作が成立せず無限ループ
そのため backend 側で問答無用に上書きする sanitizer を挟む。
_FORCED_TOOL_ARGS: dict[str, object] = {"save_file": False}
_DROPPED_TOOL_ARGS: tuple[str, ...] = ("output_dir",)
def _sanitize_tool_args(args: dict) -> dict:
for key in _DROPPED_TOOL_ARGS:
args.pop(key, None)
args.update(_FORCED_TOOL_ARGS) # LLM が指定しても必ず False で上書き
return args
プロンプト側でも save_file=false で必ず呼ぶこと と書いているが、LLM が null を返してきて upsert で打ち消されないケースがあったため、コード側強制が必要だった。プロンプトとコードの二重防御が安全。
LLM のコンテキスト長は有限 (gpt-4o で 128k トークン)。MCP の tool 結果がそのまま messages に積まれていくので、数十 KB の GeoJSON が複数返ると一発で枯渇する。tool_choice=auto でループするので 1 回でも嵩むと 2 回目以降確実に超える。
_serialize_mcp_result でツール結果ごとに文字数上限を設けて切り詰める。
_MAX_TOOL_RESULT_CHARS = 15000 # ~5000 token 程度
def _serialize_mcp_result(content: list) -> str:
texts = [c.text for c in content if getattr(c, "text", None)]
serialized = "\n".join(texts)
if len(serialized) > _MAX_TOOL_RESULT_CHARS:
serialized = (
serialized[:_MAX_TOOL_RESULT_CHARS]
+ '..."__truncated_by_char_limit__"'
)
return serialized
切り詰め時に末尾マーカー __truncated_by_char_limit__ を残しておくと、LLM が「全データ取れていない」と認識して回答に断りを入れてくれる。
MCP セッションの初期化 (session.initialize() + tools/list) は数百 ms かかる。これをリクエストごとに張り直すとレイテンシが上乗せされて辛い。FastAPI の lifespan で起動時に 1 本だけ確立し、全リクエストで共有する。
# reinfolib_service.py
_SINGLETON = ReinfolibService() # モジュールロード時に 1 つだけ作る
def get_reinfolib_service() -> ReinfolibService:
return _SINGLETON
# main.py
@asynccontextmanager
async def lifespan(_app: FastAPI):
service = None
if env.ENABLE_REINFOLIB_MCP:
service = get_reinfolib_service()
try:
await service.startup() # MCP セッション確立 + tools/list
except Exception as e:
logger.error(f"MCP 起動失敗（機能無効で起動継続）: {e}")
service = None
yield
if service is not None:
await service.shutdown()
app = FastAPI(lifespan=lifespan)
MCP 接続失敗時は機能 OFF で backend 自体は起動継続させる、という graceful degradation も入れている。MCP サイドカーが起動失敗しても、地理空間質問以外は普通に応答できる。
stg では有効化、prd では無効のまま運用したかったので、3 つの環境変数を「セットで」設定する設計にした。
| 環境変数 | 役割 |
|---|---|
| ENABLE_REINFOLIB_MCP | backend が MCP に接続するか (true/false) |
| REINFOLIB_API_KEY | 国交省 API キー |
| COMPOSE_PROFILES | docker-compose の profile 指定 (mcp でサイドカー起動) |
3 つが揃って初めて MCP が有効化される構造で、いずれか欠けても他の挙動に影響しない。さらに Intent Analyzer のプロンプト分類カテゴリも Feature Flag で切り替え:
def _build_system_prompt() -> str:
if env.ENABLE_REINFOLIB_MCP:
sections = [_BASE_PROMPT, _GEOSPATIAL_PROMPT_ADDITION]
candidates = ["person_query", "branch_query", "geospatial_query",
"knowledge_query", "chitchat"]
else:
sections = [_BASE_PROMPT] # 元のままに完全一致
candidates = ["person_query", "branch_query",
"knowledge_query", "chitchat"]
...
Flag OFF 時のプロンプトが既存実装と bit-exact で一致することは、念のためテストスクリプトで assert ORIGINAL == _build_system_prompt() を回して確認した。MCP 無効化時は既存ルート (chitchat / KB 検索) の挙動が完全に維持される。
MCP Server がどの API でデータを返すか / 返さないかを把握するため、mcp Python SDK で直接全 API を呼び出して結果を分類する検証スクリプトを書いた。
import asyncio, json
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
LAT, LON = 35.6580, 139.7016 # 渋谷駅
async def test_one(session, api_id: int):
args = {"lat": LAT, "lon": LON, "target_apis": [api_id], "save_file": False}
result = await session.call_tool("get_multi_api", args)
text = "\n".join(getattr(c, "text", "") for c in result.content)
data = json.loads(text)
api_results = data.get("data", {}).get("api_results", [])
feature_count = sum(
len(r.get("data", {}).get("features", []))
for r in api_results if isinstance(r, dict)
)
return feature_count
async def main():
async with streamablehttp_client("http://localhost:3001/mcp") as (r, w, _):
async with ClientSession(r, w) as session:
await session.initialize()
for api_id in range(1, 31):
count = await test_one(session, api_id)
print(f"API {api_id:2d} → features={count}")
これを実行して「データが返る API」「座標周辺になく 0 件の API」を分類しておくと、Intent Analyzer のプロンプトを書く時の解像度が上がる。我々のケースでは「市区町村全体の集計は取れない、点検索特化」という MCP 側の設計制約を早期に把握できた。
公式 MCP の固定 SHA pinning: α 版 / 開発初期の MCP は破壊的変更が頻繁。Dockerfile の ARG MCP_COMMIT_SHA で固定すると、サイドカーの挙動が予期せず変わるリスクを抑えられる
mcp-proxy の --pass-environment を忘れない: subprocess に環境変数を継承しないと API キーが届かず 401
Streamable HTTP のセッション管理: ClientSession は stateful なので、AsyncExitStack で __aenter__ / __aexit__ をまとめて管理する
タスク定義の dependsOn: api コンテナの dependsOn: { containerName: "sidecar", condition: "START" } を入れないと、起動順次第で MCP 接続失敗から始まる
Intent Analyzer のプロンプトと MCP のツール範囲を揃える: プロンプトが「市場データのみ」と書いてあると、施設検索や災害リスク質問が geospatial_query に分類されず MCP が使われない
→ mcp-proxy の --pass-environment 指定漏れ。コンテナ env に API キーがあっても subprocess に届いていない。
→ MCP ツールに save_file=false が渡っていない。プロンプトだけだと LLM が null で打ち消すケースがあるので、コード側 sanitizer で強制上書きする。
→ tool 結果が大きすぎる。_MAX_TOOL_RESULT_CHARS で切り詰める。一回の質問で複数 target_apis を呼ぶケースは特に注意。
→ MCP ClientSession の同時呼び出しで JSON-RPC が混線。asyncio.Lock で call_tool を直列化する。
MCP Server は AI ではなく「ツール棚」。Function Calling のオーケストレーションは自前
Claude Desktop 専用 MCP を Web サービスで使うには stdio→HTTP 化 + adaptation が必須
同時アクセス対策、ツール引数 sanitizer、結果サイズ制御、セッション維持を設計に組み込む
Feature Flag を 3 変数連動させて、Flag OFF 時に既存挙動が bit-exact 維持される構造にする
「公式 MCP を使う = 全部いい感じ」ではなく、業務向けの adaptation コードを書く前提で設計する
MCP プロトコル自体は良い抽象化を提供してくれるが、Claude Desktop / IDE 想定のエコシステムをそのまま業務サービスに持ち込むと設計のミスマッチが多い。本記事のように「サイドカー + adaptation 層」で薄くラップするパターンを 1 つ持っておくと、似たような MCP を別プロジェクトに組み込む時の再利用が効きやすい。
