内容

はじめに

背景説明

なぜ、これが重要なのか

アーキテクチャ概要

コンポーネントアーキテクチャ

1. クライアントアプリケーション層

2. MCPサーバプラットフォーム層

企業セキュリティの実装

OpenID接続認証

主な利点

実装の概要

Open Policy Agentによる詳細な許可

認可ポリシー構造

Python OPAの統合

HashiCorp Vaultを使用したセキュアシークレット管理

主な特長

実装

コアMCPサーバ構造

レガシー統合用のREST APIプロキシ

監視と監視

ELKスタックの統合

主要なモニタリングメトリック

一時的なワークフローの統合

導入と拡張性

コンテナオーケストレーション

パフォーマンスとセキュリティの考慮事項

セキュリティのベストプラクティス

パフォーマンスの最適化

メトリックの監視

パフォーマンスメトリックと結果

得られた教訓とベストプラクティス

主な成功要因

避けるべき一般的な落とし穴

今後の機能拡張

結論

著者について

参考資料

はじめに

このドキュメントでは、業界のベストプラクティスを使用して実稼働対応Model Context Protocol(MCP)サーバを構築するための包括的なリファレンスアーキテクチャについて説明します。Cisco Catalyst Center、ServiceNow、およびその他のエンタープライズシステムを統合した実際の実装を通じて実証されています。 MCPは、AIシステムが外部のサービスやデータソースと相互作用する方法のパラダイムシフトを表します。ただし、プロトタイプから実稼働への移行には、認証、許可、監視、拡張性を含むエンタープライズグレードのパターンを実装する必要があります。

背景説明

AIを活用した自動化を採用する企業が増えるにつれ、堅牢でセキュアかつスケーラブルな統合プラットフォームの必要性が高まっています。従来のポイントツーポイント統合では、メンテナンスのオーバーヘッドとセキュリティの脆弱性が生じます。Model Context Protocol(MCP)は、AIシステム統合に対する標準化されたアプローチを提供しますが、実稼働環境では、基本的なMCP実装を超えるエンタープライズグレードの機能が必要です。

この記事では、次のものを組み込んだ実稼働準備のMCPサーバプラットフォームを構築する方法について説明します。

1. エンタープライズ認証:OpenID Connect(OIDC)とCisco Duoの統合

2. きめ細かい許可：オープンポリシーエージェント(OPA)を使用したコード単位のポリシー

3. Secure Secret Management：クレデンシャルと設定用のHashiCorp Vault

4. 包括的なモニタリング：監視とトラブルシューティングのためのELKスタック

5. ワークフローオーケストレーション：複雑で長時間実行されるプロセス向けのTemporal.io

6. レガシー統合：既存システムのREST APIプロキシ

なぜ、これが重要なのか

従来の統合アプローチには、次のような制約があります。

1. セキュリティギャップ：ハードコードされたクレデンシャルと過剰な特権アクセス

2. 運用の複雑さ：分散システムの監視とトラブルシューティングが困難

3. 拡張性の問題：ポイントツーポイント統合は、増大する要件に合わせて拡張できません

4. メンテナンスのオーバーヘッド：各統合には、カスタム認証とエラー処理が必要です

エンタープライズパターンを備えたMCPアプローチは、AI主導の自動化のための標準化された再利用可能な基盤を提供しながら、これらの課題に対処します。

アーキテクチャ概要

リファレンスアーキテクチャでは、クライアントアプリケーションをMCPサーバプラットフォームから分離する階層型アプローチを実装し、複数のアプリケーションが同じエンタープライズグレードのMCPインフラストラクチャを利用できるようにします。

コンポーネントアーキテクチャ

1. クライアントアプリケーション層

クライアント層は、ユーザインターフェイスとオーケストレーションロジックを提供します。

- フロントエンド:Cisco Polarity UIフレームワークを使用した角度のあるアプリケーション

- バックエンド：ワークフローオーケストレーション用のLangGraphマルチエージェントシステム

- 認証：企業アイデンティティプロバイダーとのOIDC統合

2. MCPサーバプラットフォーム層

プラットフォーム層は、共有サービスを備えたエンタープライズグレードのMCPサーバを実装します。

コアMCPサーバ：

- mcp-catalyst-center:Ciscoネットワークデバイス管理

- mcp-service-now:ITSM統合およびチケット管理

- mcp-github：ソースコードおよびリポジトリ管理

- mcp-radkit：ネットワーク分析とモニタリング

- mcp-rest-api-proxy：レガシーシステム統合

Enterprise Services:

- 認証サービス:OIDCトークン検証およびユーザー管理

- 認証サービス:OPAベースのポリシー適用

- シークレット管理：資格情報と構成の保管をボールトベースで行います。

- <モニタリングスタック：ロギング、メトリック、アラートのELK

- ワークフローエンジン：複雑なプロセスのオーケストレーションのための一時的な機能

企業セキュリティの実装

OpenID接続認証

このプラットフォームは、OpenID Connectを使用してエンタープライズグレードの認証を実装し、Cisco Duoを通じて多要素認証をサポートしながら、既存のアイデンティティプロバイダーとのシームレスな統合を提供します。

主な利点

- シングルサインオン(SSO)：ユーザはすべてのMCPサービスで1回認証されます

- Multi-Factor Authentication:Cisco Duoの統合によりセキュリティを強化

- トークンベースのセキュリティ:JWTトークンを使用したステートレス認証

- 一元管理：既存のIdPを使用したユーザプロビジョニングおよびプロビジョニング解除

実装の概要

ファイル：mcp-common-app/src/mcp_common/oidc_auth.py

"""OIDC Authentication Module - Enterprise-grade token validation with Vault integration"""

import requests

from typing import Dict, Any, Optional

from fastapi import HTTPException

def get_oidc_config_from_vault() -> Dict[str, Any]:

"""Retrieve OIDC configuration from Vault with caching."""

vault_client = get_vault_client_with_retry()

config = vault_client.get_secret("oidc/config")

if not config:

raise ValueError("OIDC configuration not found in Vault")

# Validate required fields

required_fields = ["issuer", "client_id", "user_info_endpoint"]

missing_fields = [field for field in required_fields if field not in config]

if missing_fields:

raise ValueError(f"Missing required OIDC config fields: {missing_fields}")

return config

def verify_token_with_oidc(token: str) -> Dict[str, Any]:

"""Verify OIDC token and extract user information."""

config = get_oidc_config_from_vault()

response = requests.get(

config["user_info_endpoint"],

headers={"Authorization": f"Bearer {token}"},

timeout=10

)

if response.status_code == 200:

user_info = response.json()

if "sub" not in user_info:

raise HTTPException(status_code=401, detail="Invalid token: missing subject")

return user_info

else:

raise HTTPException(status_code=401, detail="Token validation failed")

Open Policy Agentによる詳細な許可

OPAは、柔軟なポリシーをコードとして許可し、ユーザ属性、リソースタイプ、コンテキスト情報に基づいて、きめ細かなアクセス制御を可能にします。

認可ポリシー構造

ファイル：common-services/opa/config/policy.rego

# Authorization Policy for MCP Server Platform - RBAC Implementation

package authz

default allow = false

# Administrative access - full permissions

allow {

group := input.groups[_]

group == "admin"

}

# Network engineers - Catalyst Center access

allow {

group := input.groups[_]

group == "network-engineers"

input.resource == "catalyst-center"

allowed_actions := ["read", "write", "execute"]

allowed_actions[_] == input.action

}

# Service desk - ServiceNow and read-only network access

allow {

group := input.groups[_]

group == "service-desk"

input.resource in ["servicenow", "catalyst-center"]

input.resource == "servicenow" or input.action == "read"

}

# Developers - GitHub and REST API proxy access

allow {

group := input.groups[_]

group == "developers"

input.resource in ["github", "rest-api-proxy"]

}

Python OPAの統合

<ファイル：mcp-common-app/src/mcp_common/opa.py

"""OPA Integration - Centralized authorization with audit logging"""

import os

import json

import requests

from typing import List, Dict, Any

from dataclasses import dataclass

@dataclass

class AuthorizationRequest:

"""Structure for authorization requests to OPA."""

user_groups: List[str]

resource: str

action: str

context: Dict[str, Any] = None

class OPAClient:

"""Client for interacting with Open Policy Agent (OPA) for authorization decisions."""

def __init__(self, opa_addr: str = None):

self.opa_addr = opa_addr or os.getenv("OPA_ADDR", "http://opa:8181")

self.opa_url = f"{self.opa_addr}/v1/data/authz/allow"

def check_permission(self, auth_request: AuthorizationRequest) -> bool:

"""Check if a user has permission to perform an action on a resource."""

try:

opa_input = {

"input": {

"groups": auth_request.user_groups,

"resource": auth_request.resource,

"action": auth_request.action

}

}

if auth_request.context:

opa_input["input"]["context"] = auth_request.context

response = requests.post(self.opa_url, json=opa_input, timeout=5)

if response.status_code == 200:

result = response.json()

allowed = result.get("result", False)

self._audit_log(auth_request, allowed)

return allowed

else:

print(f"OPA authorization check failed: {response.status_code}")

return False  # Fail secure

except requests.RequestException as e:

print(f"OPA connection error: {e}")

return False  # Fail secure

def _audit_log(self, auth_request: AuthorizationRequest, allowed: bool):

"""Log authorization decisions for audit purposes."""

log_entry = {

"user_groups": auth_request.user_groups,

"resource": auth_request.resource,

"action": auth_request.action,

"allowed": allowed

}

print(f"Authorization Decision: {json.dumps(log_entry)}")

# Usage decorator for MCP server methods

def require_permission(resource: str, action: str):

"""Decorator for MCP server methods that require authorization."""

def decorator(func):

async def wrapper(self, *args, **kwargs):

user_groups = getattr(self, 'user_groups', [])

if not user_groups:

raise Exception("User groups not found in request context")

opa_client = OPAClient()

auth_request = AuthorizationRequest(

user_groups=user_groups, resource=resource, action=action

)

if not opa_client.check_permission(auth_request):

raise Exception(f"Access denied for {action} on {resource}")

return await func(self, *args, **kwargs)

return wrapper

return decorator

HashiCorp Vaultを使用したセキュアシークレット管理

HashiCorp Vaultは、暗号化、アクセス制御、監査ロギングを備えたエンタープライズグレードのシークレット管理を提供します。MCPプラットフォームは、Vaultを統合して、APIクレデンシャル、データベースパスワード、設定データなどの機密情報を安全に保存および取得します。

主な特長

- 保管中および転送中の暗号化：すべてのシークレットはAES 256ビット暗号化を使用して暗号化されます

- ダイナミックシークレット：外部サービスの期限付き資格情報を生成します

- アクセス制御：きめ細かなポリシーにより、誰がどのシークレットにアクセスできるかを制御

- 監査ロギング：すべてのシークレットアクセス操作の完全な監査証跡

- シークレットローテーション：資格情報と証明書の自動ローテーション

実装

ファイル：mcp-common-app/src/mcp_common/vault.py

"""HashiCorp Vault Integration - Secure secret management with audit logging"""

import os

import json

import requests

from typing import Dict, Any, Optional, List

from datetime import datetime

class VaultClient:

"""Enterprise HashiCorp Vault client for secure secret management."""

def __init__(self, vault_addr: str = None, vault_token: str = None,

mount_point: str = "secret"):

self.vault_addr = vault_addr or os.getenv("VAULT_ADDR", "http://vault:8200")

self.vault_token = vault_token or os.getenv("VAULT_TOKEN")

self.mount_point = mount_point

self.headers = {"X-Vault-Token": self.vault_token}

if not self.vault_token:

raise ValueError("Vault token must be provided or set in VAULT_TOKEN")

def set_secret(self, path: str, secret_data: Dict[str, Any]) -> bool:

"""Store a secret in Vault KV store."""

try:

response = requests.post(

f"{self.vault_addr}/v1/{self.mount_point}/data/{path}",

headers=self.headers,

json={"data": secret_data},

timeout=10

)

success = response.status_code in [200, 204]

self._audit_log("set_secret", path, success)

return success

except requests.RequestException as e:

self._audit_log("set_secret", path, False, error=str(e))

return False

def get_secret(self, path: str) -> Optional[Dict[str, Any]]:

"""Retrieve a secret from Vault KV store."""

try:

response = requests.get(

f"{self.vault_addr}/v1/{self.mount_point}/data/{path}",

headers=self.headers,

timeout=10

)

success = response.status_code == 200

self._audit_log("get_secret", path, success)

if success:

return response.json()["data"]["data"]

return None

except requests.RequestException as e:

self._audit_log("get_secret", path, False, error=str(e))

return None

def _audit_log(self, operation: str, path: str, success: bool, error: str = None):

"""Log secret operations for audit purposes."""

log_entry = {

"timestamp": datetime.utcnow().isoformat(),

"operation": operation,

"path": f"{self.mount_point}/{path}",

"success": success

}

if error:

log_entry["error"] = error

print(f"Vault Audit: {json.dumps(log_entry)}")

# Usage mixin for MCP servers

class MCPSecretMixin:

"""Mixin class for MCP servers to easily access Vault secrets."""

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

self._vault_client = None

@property

def vault_client(self) -> VaultClient:

if self._vault_client is None:

self._vault_client = VaultClient()

return self._vault_client

def get_api_credentials(self, service_name: str) -> Optional[Dict[str, Any]]:

"""Get API credentials for a specific service."""

return self.vault_client.get_secret(f"api/{service_name}")

コアMCPサーバ構造

各MCPサーバには、エンタープライズセキュリティ統合の一貫したパターンが含まれています。

ファイル：mcp-catalyst-center/src/main.py

"""Cisco Catalyst Center MCP Server - Enterprise implementation"""

from mcp_common import VaultClient, OPAClient, get_logger, require_permission

from fastmcp import FastMCP

import os

app = FastMCP("Cisco Catalyst Center MCP Server")

logger = get_logger(__name__)

# Initialize enterprise services

vault_client = VaultClient()

opa_client = OPAClient()

@app.tool()

@require_permission("catalyst-center", "read")

async def get_all_templates(request) -> str:

"""Fetch all configuration templates from Catalyst Center."""

# Get credentials from Vault

credentials = vault_client.get_secret("api/catalyst-center")

if not credentials:

raise Exception("Catalyst Center credentials not found")

try:

# API call implementation

templates = await fetch_templates_from_api(credentials)

logger.info(f"Retrieved {len(templates)} templates")

return {

"templates": templates,

"status": "success",

"count": len(templates)

}

except Exception as e:

logger.error(f"Failed to fetch templates: {e}")

raise Exception(f"Template fetch failed: {str(e)}")

@app.tool()

@require_permission("catalyst-center", "write")

async def deploy_template(template_id: str, device_id: str) -> str:

"""Deploy configuration template to network device."""

credentials = vault_client.get_secret("api/catalyst-center")

# Implementation details...

logger.info(f"Deployed template {template_id} to device {device_id}")

return {"status": "deployed", "template_id": template_id, "device_id": device_id}

レガシー統合用のREST APIプロキシ

このプラットフォームには、非MCPクライアントをサポートするREST APIプロキシが含まれています。

ファイル：mcp-rest-api-proxy/main.py

"""REST API Proxy - Bridge between REST clients and MCP servers"""

from fastapi import FastAPI, HTTPException, Request

from langchain_mcp_adapters.client import MultiServerMCPClient

app = FastAPI()

# MCP server configurations

MCP_SERVERS = {

"servicenow": "http://mcp-servicenow:8080/mcp/",

"catalyst-center": "http://mcp-catalyst-center:8002/mcp/",

"github": "http://mcp-github:8000/mcp/"

}

client = MultiServerMCPClient({

server_name: {"url": url, "transport": "streamable_http"}

for server_name, url in MCP_SERVERS.items()

})

@app.post("/api/v1/mcp/{server_name}/tools/{tool_name}")

async def execute_tool(server_name: str, tool_name: str, request: Request):

"""Execute MCP tool via REST API for legacy clients."""

try:

body = await request.json()

result = await client.call_tool(

server_name=server_name,

tool_name=tool_name,

arguments=body.get("arguments", {})

)

return {

"status": "success",

"result": result,

"server": server_name,

"tool": tool_name

}

except Exception as e:

raise HTTPException(status_code=500, detail=f"Tool execution failed: {str(e)}")

@app.get("/api/v1/mcp/{server_name}/tools")

async def list_tools(server_name: str):

"""List available tools for a specific MCP server."""

tools = await client.list_tools(server_name)

return {"server": server_name, "tools": tools}

監視と監視

ELKスタックの統合

プラットフォームは、ELKスタックを使用して包括的なロギングを実装します。

ファイル：mcp-common-app/src/mcp_common/logger.py

"""Structured Logging for ELK Stack Integration"""

import logging

import json

from datetime import datetime

from pythonjsonlogger import jsonlogger

class StructuredLogger:

def __init__(self, name: str, level: str = "INFO"):

self.logger = logging.getLogger(name)

self.logger.setLevel(getattr(logging, level.upper()))

# JSON formatter for ELK ingestion

formatter = jsonlogger.JsonFormatter(

fmt='%(asctime)s %(name)s %(levelname)s %(message)s'

)

handler = logging.StreamHandler()

handler.setFormatter(formatter)

self.logger.addHandler(handler)

def log_mcp_call(self, tool_name: str, user: str, duration: float, status: str):

"""Log MCP tool invocation with structured data."""

self.logger.info("MCP tool executed", extra={

"tool_name": tool_name,

"user": user,

"duration_ms": duration,

"status": status,

"service_type": "mcp_server"

})

def get_logger(name: str) -> StructuredLogger:

"""Get configured logger instance."""

return StructuredLogger(name)

主要なモニタリングメトリック

このプラットフォームは、企業の運用に不可欠なメトリックを追跡します。

- 要求遅延：ツールごとの実行時間とパーセンタイル

- 認証メトリック：成功/失敗率および応答時間

- 許可の決定：ポリシー評価の頻度と結果

- シークレットアクセス：ボールト操作とクレデンシャルの使用パターン

- エラー率：サービスレベルのエラー追跡および警告しきい値

- リソース使用率：サービスあたりのCPU、メモリ、およびネットワーク使用率

一時的なワークフローの統合

複雑で長時間実行されるプロセスでは、プラットフォームはTemporal.ioを活用します。

ファイル：temporal-service/src/workflows/template_deployment.py

"""Template Deployment Workflow - Orchestrated automation with error handling"""

from temporalio import workflow, activity

from datetime import timedelta

@workflow.defn

class TemplateDeploymentWorkflow:

@workflow.run

async def run(self, deployment_request: dict) -> dict:

"""Orchestrate template deployment with proper error handling."""

# Step 1: Validate template and device

validation_result = await workflow.execute_activity(

validate_deployment, deployment_request,

start_to_close_timeout=timedelta(minutes=5)

)

if not validation_result["valid"]:

return {"status": "failed", "reason": "Validation failed"}

# Step 2: Create ServiceNow ticket

ticket_result = await workflow.execute_activity(

create_servicenow_ticket, validation_result,

start_to_close_timeout=timedelta(minutes=2)

)

# Step 3: Deploy template

deployment_result = await workflow.execute_activity(

deploy_template, {

**deployment_request,

"ticket_id": ticket_result["ticket_id"]

},

start_to_close_timeout=timedelta(minutes=30)

)

# Step 4: Close ticket

await workflow.execute_activity(

close_servicenow_ticket, {

"ticket_id": ticket_result["ticket_id"],

"deployment_result": deployment_result

},

start_to_close_timeout=timedelta(minutes=2)

)

return {

"status": "completed",

"ticket_id": ticket_result["ticket_id"],

"deployment_id": deployment_result["deployment_id"]

}

@activity.defn

async def validate_deployment(request: dict) -> dict:

"""Validate deployment request against business rules."""

# Validation logic implementation

return {"valid": True, "validated_request": request}

@activity.defn

async def deploy_template(request: dict) -> dict:

"""Execute template deployment via Catalyst Center."""

# Template deployment logic

return {"deployment_id": "deploy_123", "status": "success"}

導入と拡張性

コンテナオーケストレーション

このプラットフォームでは、開発にDocker Composeを使用します。Kubernetesは実稼働に使用できます。

ファイル：docker-compose.yml（抜粋）

version: '3.8'

services:

mcp-catalyst-center:

build: ./mcp-catalyst-center

environment:

- VAULT_ADDR=http://vault:8200

- OPA_ADDR=http://opa:8181

- ELASTICSEARCH_URL=http://elasticsearch:9200

depends_on: [vault, opa, elasticsearch]

networks: [mcp-network]

vault:

image: hashicorp/vault:latest

environment:

VAULT_DEV_ROOT_TOKEN_ID: myroot

VAULT_DEV_LISTEN_ADDRESS: 0.0.0.0:8200

cap_add: [IPC_LOCK]

networks: [mcp-network]

opa:

image: openpolicyagent/opa:latest-envoy

command: ["run", "--server", "--config-file=/config/config.yaml", "/policies"]

volumes: ["./common-services/opa/config:/policies"]

networks: [mcp-network]

パフォーマンスとセキュリティの考慮事項

セキュリティのベストプラクティス

1. ゼロトラストアーキテクチャ：すべての要求が認証および許可されます

2. シークレット回転:Vaultを使用したシークレット回転の自動化

3. ネットワークのセグメント化:mTLSによるサービスメッシュ

4. 監査ロギング:ELKの包括的な監査証跡

パフォーマンスの最適化

1. 接続プーリング：外部APIへのHTTP接続を再利用

2. キャッシング：頻繁にアクセスするデータのRedisベースのキャッシング

3. 非同期処理：スタック全体でノンブロッキングI/Oを実現

4. ロードバランシング：複数のMCPサーバインスタンス間で負荷を分散

メトリックの監視

主なメトリックは次のとおりです。

- MCPツールごとの要求遅延

- 認証の成功/失敗率

- リソースごとの承認の決定

- 秘密の取得頻度

- サービスコンポーネント別のエラー率

パフォーマンスメトリックと結果

パフォーマンステスト中、プラットフォームは以下を達成しました。

- 認証の決定に100ミリ秒未満の遅延

- すべてのMCPサービスで99.9 %のアップタイム

- 最大1000人の同時ユーザに対応する直線的な拡張性

- 統合開発時間を90 %短縮

得られた教訓とベストプラクティス

主な成功要因

1. 標準化：共通ライブラリにより重複やバグが減少

2. 可観測性：包括的なロギングによる迅速なトラブルシューティング

3. 設計によるセキュリティ：初日から認証と許可

4. モジュール性：独立したMCPサーバにより、対象を絞った拡張が可能

5. Documentation：明確なAPIドキュメントにより導入を加速

避けるべき一般的な落とし穴

1. オーバーエンジニアリング：シンプルな構成から始め、必要に応じて複雑さを増やす

2. 緊密な連携:MCPサーバの疎結合を維持

3. セキュリティの後付け：最初からセキュリティを構築する

4. 不十分なテスト：包括的なテスト戦略の実施

5. 不適切なエラー処理：グレースフルなデグレードを保証

今後の機能拡張

プラットフォームロードマップには次のものが含まれます。

1. AI-Driven Insights:MLベースの異常検出

2. マルチクラウドサポート：クラウドプロバイダー間の導入

3. Workflow Marketplace：再利用可能なワークフローテンプレート

4. 高度な分析：リアルタイムダッシュボードとレポート

結論

実稼働グレードのMCPサーバを構築するには、セキュリティ、スケーラビリティ、モニタリング、メンテナンス性などのエンタープライズ要件を慎重に検討する必要があります。このリファレンスアーキテクチャでは、業界標準のツールとパターンを使用してこれらの機能を実装する方法を示します。

モジュラ設計により、組織はMCPを徐々に導入しながら、エンタープライズクラスのセキュリティと運用を初日から保証できます。OIDC、OPA、Vault、ELKなどの実績のあるテクノロジーを活用することで、チームはインフラストラクチャの問題ではなく、ビジネスロジックに集中できます。

著者について

この記事は、シスコの社内イノベーションイニシアチブの一環としてMCP Fusionersチームによって開発され、エンタープライズAIシステム統合に対する実用的なアプローチを実証しています。

参考資料

1. Model Context Protocol Specification:https://modelcontextprotocol.io/

2. OpenID Connect Core 1.0:https://openid.net/specs/openid-connect-core-1_0.html

3. Open Policy Agentのドキュメント – https://www.openpolicyagent.org/docs

4. HashiCorp Vaultドキュメント – https://www.vaultproject.io/docs

5. Temporal.ioドキュメント – https://docs.temporal.io/

6. ELKスタックガイド – https://www.elastic.co/elastic-stack/