目錄

簡介

背景資訊

這為什麼重要

架構概觀

元件體系結構

1.客戶端應用層

2. MCP伺服器平台層

企業安全實施

OpenID連線驗證

主要優勢

實施概述

使用開放式策略代理的細粒度授權

授權策略結構

Python OPA整合

使用HashiCorp Vault進行安全金鑰管理

主要功能

實現

核心MCP伺服器結構

適用於舊版整合的REST API代理

監控和可觀察性

ELK堆疊整合

關鍵監控指標

臨時工作流整合

部署和可擴充性

容器協調

效能和安全注意事項

安全最佳實踐

效能最佳化

監控指標

效能度量和結果

經驗教訓和最佳做法

關鍵成功因素

要避免的常見陷阱

未來的增強功能

結論

作者簡介

參考資料

簡介

本文檔介紹使用行業最佳實踐構建生產就緒型模型上下文協定(MCP)伺服器的綜合參考體系結構，該體系結構通過整合Cisco Catalyst Center、ServiceNow和其他企業系統的真實實施進行了演示。MCP代表了AI系統與外部服務和資料來源互動方式的正規化轉變。但是，從原型過渡到生產需要實施企業級模式，包括身份驗證、授權、監控和可擴充性。

背景資訊

隨著組織越來越多地採用AI驅動的自動化，對穩健、安全且可擴展的整合平台的需求變得至關重要。傳統的點對點整合會產生維護開銷和安全漏洞。模型上下文協定(MCP)為AI系統整合提供了標準化的方法，但生產部署需要超越基本MCP實施的企業級功能。

本文演示了如何構建生產就緒型MCP伺服器平台，該平台包含以下內容：

1. 企業驗證:OpenID Connect(OIDC)與Cisco Duo整合

2. 精細授權:使用開放式策略代理(OPA)的Policy-as-code

3. 安全金鑰管理:用於憑證和配置的HashiCorp Vault

4. 全面監控:用於觀察和故障排除的ELK堆疊

5. 工作流協調：適用於複雜、長期運行的進程的temporal.io

6. 傳統整合:適用於現有系統的REST API代理

這為什麼重要

傳統整合方法存在以下幾個侷限性：

1. 安全差距:硬編碼憑證和超許可權訪問

2. 操作複雜性：難以對分散式系統進行監控和故障排除

3. 可擴充性問題:點對點整合無法隨需求的增長而擴展

4. 維護開銷:每個整合都需要自定義身份驗證和錯誤處理

採用企業模式的MCP方法可以解決這些挑戰，同時為AI驅動的自動化提供一個標準化、可重複使用的基礎。

架構概觀

參考體系結構實施了一種分層的方法，將客戶端應用程式與MCP伺服器平台分開，使多個應用程式能夠利用相同的企業級MCP基礎架構。

元件體系結構

1.客戶端應用層

客戶端層提供使用者介面和協調邏輯：

- 前端:使用Cisco Polarity UI框架的角度應用程式

- 後端:工作流協調的LangGraph多Agent系統

- 驗證:OIDC與企業身份提供商整合

2. MCP伺服器平台層

平台層通過共用服務實現企業級MCP伺服器：

核心MCP伺服器：

- mcp-catalyst-center:思科網路裝置管理

- mcp-service-now:ITSM整合和票證管理

- mcp-github:原始碼和儲存庫管理

- mcp-radkit:網路分析與監控

- mcp-rest-api-proxy:舊版系統整合

企業服務：

- 驗證服務:OIDC令牌驗證和使用者管理

- 授權服務:基於OPA的策略實施

- 金鑰管理:基於保管庫的憑證和配置儲存

- <Monitoring Stack:用於記錄、度量和警報的ELK

- 工作流引擎:複雜流程協調的時間性

企業安全實施

OpenID連線驗證

該平台使用OpenID Connect實施企業級身份驗證，提供與現有身份提供商的無縫整合，同時支援通過Cisco Duo進行多重身份驗證。

主要優勢

- 單一登入(SSO):使用者在所有MCP服務之間驗證一次

- 多重驗證:整合的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")

使用開放式策略代理的細粒度授權

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整合

<File: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代理

該平台包括一個REST API代理以支援非MCP客戶端：

檔案：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)

關鍵監控指標

該平台跟蹤企業運營的基本指標：

- 請求延遲:每個工具的執行時間和百分比

- 身份驗證指標：成功/失敗率和響應時間

- 授權決策:政策評價頻率和成果

- Secret Access:保管庫操作和憑據使用模式

- 錯誤率:服務級別錯誤跟蹤和警報閾值

- 資源利用率:每個服務的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工具的請求延遲

- 身份驗證成功/失敗率

- 每個資源的授權決策

- 秘密檢索頻率

- 按服務元件統計的錯誤率

效能度量和結果

在效能測試過程中，平台實現了：

- 用於身份驗證決策的延遲低於100ms

- 所有MCP服務的99.9%正常運行時間

- 線性可擴展性多達1000個併發使用者

- 整合開發時間縮短了90%

經驗教訓和最佳做法

關鍵成功因素

1. 標準化:通用庫可減少重複和錯誤

2. 可觀察性:全面的日誌記錄可實現快速故障排除

3. 設計安全性:從第一天開始進行身份驗證和授權

4. 模組化:獨立的MCP伺服器可實現目標擴展

5. 說明文件:清除API文檔加快了採用速度

要避免的常見陷阱

1. 過度工程:開始簡單，根據需要增加複雜性

2. 緊耦合:保持MCP伺服器鬆散連線

3. 安全事後:從一開始就構建安全性

4. 測試不充分:實施全面的測試策略

5. 錯誤處理不佳:確保平穩降級

未來的增強功能

該平台規劃圖包括：

1. AI驅動的洞察:基於ML的異常檢測

2. 多雲支援:跨雲提供商部署

3. 工作流市場:可重複使用的工作流模板

4. 高級分析:即時控制板和報告

結論

構建生產級MCP伺服器需要仔細考慮企業需求，包括安全性、可擴充性、監控性和可維護性。此參考體系結構演示了如何使用行業標準工具和模式實施這些功能。

模組化設計使組織能夠逐步採用MCP，同時從第一天起就確保企業級安全性和操作。通過利用OIDC、OPA、Vault和ELK等成熟技術，團隊可以專注於業務邏輯而非基礎設施問題。

作者簡介

本文由MCP Fusioners團隊開發，作為思科內部創新計畫的一部分，展示企業AI系統整合的實用方法。

參考資料

1. 模型上下文協定規範 — https://modelcontextprotocol.io/

2. OpenID連線核心1.0 - https://openid.net/specs/openid-connect-core-1_0.html

3. 開啟策略代理文檔 — 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/