公開したくない社内サービスに Managed Agents を繋ぐ — MCP トンネル設計の勘所

# inventory_mcp.py — 社内ネットワーク内でのみ動かす MCP サーバー
from mcp.server.fastmcp import FastMCP
import os
import asyncpg
 
mcp = FastMCP("inventory")
 
# 接続先はすべて社内アドレス。公開 IP を一切持たない
DB_DSN = os.environ["INTERNAL_DB_DSN"]  # 例: postgres://app@10.0.3.12:5432/inventory
 
@mcp.tool()
async def check_stock(sku: str) -> dict:
    """指定 SKU の在庫数と次回入荷予定を返す。"""
    if not sku.isalnum():
        # ツール側で入力を必ず検証する。エージェントの出力を信用しない
        raise ValueError("sku は英数字のみ受け付けます")
 
    conn = await asyncpg.connect(DB_DSN)
    try:
        row = await conn.fetchrow(
            "SELECT quantity, restock_date FROM stock WHERE sku = $1",
            sku,
        )
    finally:
        await conn.close()
 
    if row is None:
        return {"sku": sku, "found": False}
    return {
        "sku": sku,
        "found": True,
        "quantity": row["quantity"],
        "restock_date": row["restock_date"].isoformat() if row["restock_date"] else None,
    }
 
if __name__ == "__main__":
    mcp.run()

# 社内ホストで実行。インバウンドのポートは一切開けない
export CLAUDE_TUNNEL_TOKEN="YOUR_TUNNEL_TOKEN"
claude-tunnel connect \
  --target stdio:./inventory_mcp.py \
  --name internal-inventory
# → 割り当てられた tunnel id (例: tnl_internal_inventory) が表示される

# agent_config.py — エージェント側。社内の詳細は何も持たない
agent = client.beta.agents.create(
    model="claude-opus-4-8",
    name="inventory-assistant",
    instructions=(
        "在庫の問い合わせには check_stock ツールを使う。"
        "ツールが found=false を返した場合は、推測で在庫数を答えない。"
    ),
    mcp_servers=[
        {
            "type": "tunnel",
            "tunnel_id": "tnl_internal_inventory",
            "tool_allowlist": ["check_stock"],  # 使えるツールを明示的に絞る
        }
    ],
    sandbox={"type": "self_hosted", "id": "sbx_team_default"},
)

# inventory_mcp.py に認可を足す
from mcp.server.fastmcp import Context
 
@mcp.tool()
async def check_stock(sku: str, ctx: Context) -> dict:
    # トンネル経由で渡るメタデータからテナントを取り出す
    tenant = ctx.request_context.meta.get("tenant")
    if tenant is None:
        raise PermissionError("テナント情報のない呼び出しは拒否します")
 
    conn = await asyncpg.connect(DB_DSN)
    try:
        row = await conn.fetchrow(
            "SELECT quantity, restock_date FROM stock "
            "WHERE sku = $1 AND tenant_id = $2",  # テナント境界を SQL で強制
            sku, tenant,
        )
    finally:
        await conn.close()
    # 以降は前述と同じ

# トンネルクライアントの薄いラッパ。指数バックオフで再接続する
import asyncio
 
async def run_tunnel_with_retry():
    backoff = 1
    while True:
        try:
            await start_tunnel(target="stdio:./inventory_mcp.py",
                               name="internal-inventory")
            backoff = 1  # 接続成功でリセット
        except TunnelDisconnected as e:
            # 切断は異常ではなく想定内のイベントとして扱う
            wait = min(backoff, 30)
            log.warning("tunnel disconnected: %s — %ds 後に再接続", e, wait)
            await asyncio.sleep(wait)
            backoff *= 2