私は以前/ECサイトのAIカスタマーサービスシステムを構築していた際、LangGraphで外部APIを呼び出すたびに「認証情報をどう管理すべきか」という壁にぶつかりました。社内CRMへのアクセス、配送追跡システムの統合、カスタマーサポートチケットの作成——これらをすべてLangGraphのTool Callingで実現したい。しかし、ベンダーAPIのキーがソースコードに露出している现状は明らかに危険です。本稿では、MCP（Model Context Protocol）ゲートウェイの役割と、代替アーキテクチャの選択肢を実務観点から整理します。
なぜLangGraphのTool Callingは企業環境では危険なのか
LangGraphはAgentアーキテクチャの実装に優れていますが、標準設定のまま外部ツールを呼び出す場合、いくつかの本質的課題が存在します。
典型的な脆弱性の例
# ❌ 危険: APIキーが直接露出
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4.1",
api_key="sk-xxxxx...", # APIキーがここに露出
base_url="https://api.holysheep.ai/v1" # HolySheep AI使用
)
tools = [
{
"name": "query_inventory",
"description": "在庫確認",
"function": lambda product_id: requests.get(
f"https://internal-corp-api.com/inventory/{product_id}",
headers={"Authorization": "Bearer internal-key-xxx"}
).json()
}
]
agent = create_react_agent(llm, tools)
LLMがプロンプトインジェクション攻撃を受けると、内部APIキーが流出する可能性
上記のコードでは、LLMへのプロンプトインジェクション攻撃成功時、認証情報が外部に漏れるリスクがあります。また、複数の社内ツール调用時に認証情報を分散管理することは運用の複雑さを増大させます。
MCPゲートウェイを導入する3つのパターン
MCPゲートウェイは、LangGraph Agentと企業内部ツールの間に立つ「認証・認可の門番」として機能します。導入パターンは大きく3つに分類されます。
パターン1: MCPサーバーとしてのLangServe
# ✅ MCPゲートウェイ兼任: LangServeで 안전한 ツール公開
server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langserve import add_routes
from langchain_core.tools import tool
from langchain_holysheep import HolySheepChatOpenAI # 成本節約
import os
app = FastAPI(title="企業ツールMCPゲートウェイ")
環境変数で認証情報を管理（Git等にはコミットしない）
CORP_API_KEY = os.environ["CORP_API_KEY"]
HOLYSHEEP_API_KEY = os.environ["HOLYSHEEP_API_KEY"]
llm = HolySheepChatOpenAI(
model="deepseek-chat",
api_key=HOLYSHEEP_API_KEY,
base_url="https://api.holysheep.ai/v1",
temperature=0.3
)
@tool
def query_inventory(product_id: str, warehouse: str = "main") -> dict:
"""企業の在庫管理系统にクエリを送信（認証済み）"""
import requests
response = requests.get(
f"https://internal-corp-api.com/inventory/{product_id}",
params={"warehouse": warehouse},
headers={
"Authorization": f"Bearer {CORP_API_KEY}",
"X-Request-Source": "mcp-gateway"
},
timeout=5
)
if response.status_code != 200:
raise HTTPException(status_code=response.status_code, detail="在庫APIエラー")
return response.json()
@tool
def create_support_ticket(customer_id: str, issue: str, priority: str = "normal") -> dict:
"""サポートチケットをCRMに作成（監査ログ出力）"""
import requests
import logging
logging.info(f"[AUDIT] ticket_create: customer={customer_id}, priority={priority}")
response = requests.post(
"https://internal-corp-api.com/tickets",
json={"customer_id": customer_id, "issue": issue, "priority": priority},
headers={"Authorization": f"Bearer {CORP_API_KEY}"}
)
return response.json()
LangGraph AgentをMCPツールとして公開
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, [query_inventory, create_support_ticket])
add_routes(app, agent, path="/agent", enabled_endpoints=["invoke", "stream"])
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8080)
このパターンでは、LangServe経由でLangGraph AgentをMCPプロトコル対応のツールとして公開します。認証情報（CORP_API_KEY）はサーバー側で環境変数として管理するため、LLMに露出しません。
パターン2: 専用MCPゲートウェイ（Recommended for Enterprise）
# ✅ 専用MCPゲートウェイ: 認証・認可・レート制限を集中管理
mcp_gateway.py
from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel
from typing import Optional
import hashlib
import hmac
import time
app = FastAPI(title="Enterprise MCP Gateway")
事前共有鍵によるリクエスト署名（LLMへの直接アクセスを 차단）
SHARED_SECRET = os.environ["MCP_SHARED_SECRET"]
class ToolRequest(BaseModel):
tool_name: str
parameters: dict
session_id: str
user_context: Optional[dict] = None
@app.post("/mcp/invoke")
async def invoke_tool(
request: ToolRequest,
authorization: str = Header(None)
):
# HMAC署名検証
if not verify_signature(request, authorization):
raise HTTPException(status_code=401, detail="無効な署名")
# レート制限チェック（HolySheep AI ¥1=$1 で成本最適化）
if not check_rate_limit(request.session_id):
raise HTTPException(status_code=429, detail="レート制限超過")
# ツールディスパッチ（権限チェック付き）
allowed_tools = get_user_permissions(request.session_id)
if request.tool_name not in allowed_tools:
raise HTTPException(status_code=403, detail="ツール利用権限なし")
# 監査ログ
log_tool_usage(request)
return await dispatch_tool(request.tool_name, request.parameters)
def verify_signature(req: ToolRequest, auth: str) -> bool:
"""リクエスト改ざん検知"""
expected = hmac.new(
SHARED_SECRET.encode(),
f"{req.session_id}:{req.tool_name}:{time.time()//60}".encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(auth, f"Bearer {expected}")
クライアントサイド: HolySheep AI LangGraph Agent
client_agent.py
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import create_react_agent
from langchain_holysheep import HolySheepChatOpenAI
llm = HolySheepChatOpenAI(
model="gpt-4.1", # $8/MTok（他のベンダー比85%節約）
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1"
)
def mcp_tool(tool_name: str, parameters: dict):
"""MCPゲートウェイ経由でツール呼び出し"""
import httpx
import hmac
import time
session = parameters.pop("_session_id", "default")
payload = {"tool_name": tool_name, "parameters": parameters, "session_id": session}
# 署名生成
signature = hmac.new(
os.environ["MCP_SHARED_SECRET"].encode(),
f"{session}:{tool_name}:{time.time()//60}".encode(),
hashlib.sha256
).hexdigest()
response = httpx.post(
"https://mcp-gateway.internal.com/mcp/invoke",
json=payload,
headers={"Authorization": f"Bearer {signature}"},
timeout=30
)
return response.json()
MCPツール定義
inventory_tool = StructuredTool.from_function(
func=lambda product_id, warehouse="main": mcp_tool("query_inventory", {"product_id": product_id, "warehouse": warehouse}),
name="query_inventory",
description="社内在庫システムを安全にクエリ"
)
ticket_tool = StructuredTool.from_function(
func=lambda customer_id, issue, priority="normal": mcp_tool("create_support_ticket", {"customer_id": customer_id, "issue": issue, "priority": priority}),
name="create_support_ticket",
description="CRMにサポートチケットを作成"
)
agent = create_react_agent(llm, [inventory_tool, ticket_tool])
HolySheep AIをLangGraph Enterprise環境に統合するメリット
私は複数の本番環境を運用していますが、HolySheep AIへの登録後はコスト構造が大きく改善されました。具体的な数値を示すと、GPT-4.1は$8/MTokですがDeepSeek V3.2は$0.42/MTokと80%以上低成本です。ECサイトのAI客服で月500万トークンを処理する場合、月額約$4,000が$200近くに削減されます。
また、WeChat Pay・Alipayに対応しているため、中国本土の開発チームとの決済もスムーズです。登録時に無料クレジットが付与されるのも、試運用期間に便利です。レイテンシは50ms未満を保証しており、リアルタイム性が求められる客服シナリオでも安心感があります。
MCPゲートウェイ不要の場合：いつ安全と判断できるか
MCPゲートウェイは銀の弾丸ではありません。以下の場合、導入コストに見合わない可能性があります。
- 全てのパートナーがSOC2認証済み：Google Cloud、AWS等のエンタープライズAPIは Already 認証情報を管理してくれる
- LangGraphが直接Internet-facingでない：VPC内のみアクセス可能
- ツール呼び出しがRead-only：データ変更権限がない場合はリスクが限定的
# ✅ Read-only外部APIのみの場合：MCP不要で安全的
from langgraph.prebuilt import create_react_agent
from langchain_holysheep import HolySheepChatOpenAI
llm = HolySheepChatOpenAI(
model="gemini-2.5-flash", # $2.50/MTok、成本重視
api_key=os.environ["HOLYSHEEP_API_KEY"],
base_url="https://api.holysheep.ai/v1"
)
読み取り専用のPublic APIのみ呼び出す場合
public_tools = [
{
"name": "get_weather",
"description": "天気情報取得",
"function": lambda city: requests.get(
f"https://api.open-meteo.com/v1/forecast",
params={"city": city, "current_weather": True}
).json()
},
{
"name": "get_exchange_rate",
"description": "為替レート取得",
"function": lambda from_cur, to_cur: requests.get(
f"https://api.exchangerate.host/latest",
params={"base": from_cur, "symbols": to_cur}
).json()
}
]
agent = create_react_agent(llm, public_tools)
認証情報不要、LLMへの露出リスクも低い
よくあるエラーと対処法
エラー1: MCP_gateway_timeout - ツール呼び出しがタイムアウトする
# 問題: 内部APIの応答遅延がLangGraphのデフォルトtimeout超過
LangGraphデフォルト: 通常30秒程度
解決: timeout設定とリトライロジックを追加
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def resilient_tool_call(tool_name: str, params: dict):
try:
async with httpx.AsyncClient() as client:
response = await client.post(
f"https://mcp-gateway.internal.com/mcp/invoke",
json={"tool_name": tool_name, "parameters": params, "session_id": "default"},
timeout=httpx.Timeout(60.0, connect=10.0) # 合計60秒timeout
)
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
# フォールバック: 直接API呼び出し（認証情報をServer Sideで保持）
return await fallback_direct_call(tool_name, params)
エラー2: authentication_realm_mismatch - 認証Realmの不一致
# 問題: LangGraph AgentとMCP Gatewayで認証情報が異なる
症状: 401 Unauthorized が不定期発生
解決: セッションキーを統一管理
import redis
redis_client = redis.from_url(os.environ["REDIS_URL"])
def get_auth_token(session_id: str) -> str:
"""Redisから有効な認証トークンを取得"""
token = redis_client.get(f"auth:{session_id}")
if token:
return token.decode()
# 新規トークン発行（OAuth2 Client Credentials Flow）
token = acquire_oauth_token(
client_id=os.environ["CORP_CLIENT_ID"],
client_secret=os.environ["CORP_CLIENT_SECRET"],
token_url="https://auth.internal.com/oauth/token"
)
redis_client.setex(f"auth:{session_id}", 3600, token) # 1時間有効
return token
Agent初期化時にセッションを共有
agent = create_react_agent(llm, tools, config={"configurable": {"session_id": session_id}})
エラー3: circular_dependency_detected - LangGraph内で無限ループ
# 問題: あるツールが別のツールを呼び出し、それが元のツールを呼び出す
例: query_inventory → check_stock_level → query_inventory → ...
解決: ツール呼び出し深度を制限
MAX_TOOL_CALL_DEPTH = 5
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict
class AgentState(TypedDict):
messages: list
tool_call_depth: int
def should_continue(state: AgentState) -> str:
if state["tool_call_depth"] >= MAX_TOOL_CALL_DEPTH:
return END
return "tools"
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge("__start__", "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")
compiled = graph.compile()
深度超過時は警告とともに処理終了
if state["tool_call_depth"] >= MAX_TOOL_CALL_DEPTH:
logger.warning(f"[SECURITY] ツール呼び出し深度超過: {state['tool_call_depth']}")
return "ツール呼び出し上限に達しました。管理者にお問い合わせください。"
エラー4: prompt_injection_detected - インジェクション攻撃
# 問題: ユーザーがプロンプト内に悪意あるコマンドを挿入
解決: MCPゲートウェイで入力サニタイズ
import re
@app.post("/mcp/invoke")
async def safe_invoke(request: ToolRequest):
# パラメータ内の悪意あるパターンを検出してブロック
dangerous_patterns = [
r"ignore previous instructions",
r"ignore all previous",
r"disregard.*instruction",
r"--ignore",
r"&#",
r"