MCP Forge

🚀 使用 FastAPI 和 FastAPI-MCP 開發 MCP 服務器框架

本項目藉助 FastAPI 和 FastAPI-MCP 構建 MCP 服務器框架，可在多個雲平臺協調管理計算任務，提供豐富 API 接口，支持多協議與集成方案，適用於複雜雲環境。

🚀 快速開始

要開始使用 FastAPI 和 FastAPI-MCP 創建一個基本的 MCP 服務器，請按照以下步驟操作：

1. 安裝必要的依賴項：

   pip install fastapi fastapi_mcp uvicorn
   

2. 創建主應用文件 (main.py)：

   from fastapi import FastAPI
   from fastapi_mcp import FastApiMCP
   
   app = FastAPI()
   
   # 定義一個標準的 FastAPI 端點
   @app.post("/predict", operation_id="predict_sentiment")
   async def predict_sentiment(text: str):
       return {"sentiment": "positive", "confidence": 0.92}
   
   # 創建並掛載 MCP 服務，自動將上面的端點轉換為 MCP 工具
   mcp = FastApiMCP(
       app,
       name="sentiment-analysis",
       description="情感分析服務",
       base_url="http://localhost:5000",
       include_operations=["predict_sentiment"]
   )
   
   # 掛載 MCP 服務到指定路徑
   mcp.mount(mount_path="/mcp")
   
   if __name__ == "__main__":
       uvicorn.run(app, host="0.0.0.0", port=8000)
   

3. 運行應用：

   python main.py
   

4. 訪問服務：

   • 訪問 FastAPI 端點：http://localhost:5000/predict
   • 訪問 MCP 工具：http://localhost:5000/mcp

✨ 主要特性

核心功能

• 跨雲平臺兼容性：支持 AWS、GCP、Azure 等主流雲服務提供商。
• 負載均衡與故障轉移：自動分配任務到不同的雲實例，確保高可用性。
• 日誌與監控集成：內置對 Prometheus 和 Grafana 的支持，方便即時監控和分析。
• 安全認證：支持 OAuth2、JWT 等多種身份驗證機制，保障數據安全。

高級功能

• 分佈式事務管理：確保跨雲服務的事務一致性。
• 緩存與加速：集成 Redis 或 Memcached 進行結果緩存，提升響應速度。
• 動態擴展：根據負載自動調整資源使用，優化成本。
• 智能路由：基於地理位置或策略分配請求到最佳可用節點。

📦 安裝指南

環境要求

• 操作系統：Linux (推薦) 或 macOS
• Python 版本：3.6+
• 依賴管理工具：poetry/pipenv (可選)
• 額外工具：Docker, kubectl (如需容器化部署)

安裝步驟

1. 克隆項目倉庫：

   git clone https://github.com/yourusername/mcp-server.git
   cd mcp-server
   

2. 創建虛擬環境並安裝依賴：

   python -m venv .venv
   source .venv/bin/activate  # 在 macOS/Linux 下
   pip install -r requirements.txt
   

3. 配置項目：

   • 修改 config.py 中的雲服務提供商憑證和 API 端點。
   • 更新日誌記錄配置，確保輸出到合適的位置。

4. 啟動服務器：

   uvicorn main:app --host 0.0.0.0 --port 8000
   

另一種安裝方式

安裝依賴

pip install mcp-server

配置文件

創建 config.yaml 文件，如下所示：

cloud_providers:
  - name: AWS
    region: us-east-2
    access_key: YOUR_AWS_KEY
    secret_key: YOUR_AWS_SECRET

  - name: GCP
    region: europe-west1
    project_id: your-gcp-project
    key_path: path/to/service-account.json

  - name: Azure
    region: eastus
    subscription_id: your_azure_subscription_id
    client_id: your_azure_client_id
    client_secret: your_azure_client_secret

啟動服務器

python3 -m mcp_server.run --config config.yaml

💻 使用示例

基礎用法

情感分析服務

# app/endpoints/sentiment_endpoints.py
from fastapi import APIRouter, Depends
from app.services.sentiment_service import SentimentService

router = APIRouter()

@router.post("/sentiment")
async def analyze_sentiment(text: str, service: SentimentService = Depends(SentimentService)):
    sentiment = await service.analyze(text)
    return {"sentiment": sentiment}

停車場搜索服務

# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService

router = APIRouter()

@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
                              service: ParkingService = Depends(ParkingService)):
    result = await service.find_nearby_parkings(latitude, longitude, radius)
    return {"data": result}

高級用法

個性化推薦系統

# app/endpoints/recommendation_endpoints.py
from fastapi import APIRouter, Depends
from app.services.recommendation_service import RecommendationService

router = APIRouter()

@router.post("/recommendations")
async def get_recommendations(user_id: int, service: RecommendationService = Depends(RecommendationService)):
    recommendations = await service.get Recommendations(user_id)
    return {"recommendations": recommendations}

分佈式事務管理

from mcp_server.transaction import TransactionManager

tm = TransactionManager()
with tm.context():
    # 執行跨雲操作
    pass

智能路由與負載均衡

class SmartRouter:
    def route_request(self, request):
        # 返回目標雲服務提供商和區域
        pass

動態擴展

# Kubernetes Horizontal Pod Autoscaler 配置
apiVersion: autoscaling/v1
kind: HorizontalPodAutoscaler
metadata:
  name: mcp-server-hpa
spec:
  scaleRef:
    kind: Deployment
    name: mcp-server-deployment
    apiVersion: apps/v1
  minReplicas: 2
  maxReplicas: 5
  targetCPUUtilizationPercentage: 80

📚 詳細文檔

項目結構

一個典型的 MCP 服務器項目可能包含以下文件和目錄：

mcp_server/
├── main.py                # 主應用文件
├── app/                  # FastAPI 應用模塊
│   ├── models.py         # 數據模型定義
│   ├── services/        # 服務類（如數據庫連接、業務邏輯）
│   └── endpoints/       # API 端點實現
├── config/               # 配置管理
│   ├── settings.py      # 應用設置
│   └── environment.py   # 環境變量管理
└── tests/                # 測試用例
    ├── test_endpoints.py  # 單元測試
    └── integration_tests.py  # 集成測試

基礎功能實現

創建服務類

# app/services/parking_service.py
class ParkingService:
    def __init__(self, db):
        self.db = db

    async def find_nearby_parkings(self, latitude: float, longitude: float, radius: int) -> list:
        # 查詢數據庫以獲取附近的停車場
        pass

實現 API 端點

# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService
from config.environment import get_db

router = APIRouter()

@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
                              service: ParkingService = Depends(ParkingService)):
    db = next(get_db())
    result = await service.find_nearby_parkings(latitude, longitude, radius)
    return {"data": result}

配置管理

# config/settings.py
class Settings:
    def __init__(self):
        self.debug = True  # 調試模式，默認為真

    @property
    def database_url(self) -> str:
        return "sqlite:///./test.db"

    @property
    def secret_key(self) -> str:
        return "your-secret-key-here"

高級功能實現

依賴注入

FastAPI 的依賴注入機制非常強大，可以用於管理各種依賴項。

# config/database.py
def get_db():
    db = connect_to_db()
    try:
        yield db
    finally:
        db.close()

@app.get("/items/")
async def get_items(db=Depends(get_db)):
    return db.query(Item).all()

環境適應

# app/services/database_service.py
class DatabaseService:
    def __init__(self, settings):
        self.settings = settings
        self.db = connect_to_db(settings.database_url)

    async def query(self, sql: str) -> list:
        pass

日誌記錄和監控

集成日誌庫（如 logging）和監控工具（如 Prometheus、Grafana）可以提供更詳細的運行時信息。

常用功能示例

情感分析服務

# app/endpoints/sentiment_endpoints.py
from fastapi import APIRouter, Depends
from app.services.sentiment_service import SentimentService

router = APIRouter()

@router.post("/sentiment")
async def analyze_sentiment(text: str, service: SentimentService = Depends(SentimentService)):
    sentiment = await service.analyze(text)
    return {"sentiment": sentiment}

停車場搜索服務

# app/endpoints/parking_endpoints.py
from fastapi import APIRouter, Depends
from app.services.parking_service import ParkingService

router = APIRouter()

@router.get("/parking/nearby")
async def find_nearby_parkings(latitude: float, longitude: float, radius: int,
                              service: ParkingService = Depends(ParkingService)):
    result = await service.find_nearby_parkings(latitude, longitude, radius)
    return {"data": result}

個性化推薦系統

# app/endpoints/recommendation_endpoints.py
from fastapi import APIRouter, Depends
from app.services.recommendation_service import RecommendationService

router = APIRouter()

@router.post("/recommendations")
async def get_recommendations(user_id: int, service: RecommendationService = Depends(RecommendationService)):
    recommendations = await service.get Recommendations(user_id)
    return {"recommendations": recommendations}

常見問題與故障排除

• 問：如何處理 API 端點的錯誤？ 答：可以使用 FastAPI 的異常處理機制，定義自定義異常並將其註冊到應用中。
• 問：如何實現速率限制和認證？ 答：可以集成第三方庫如 fastapi-limiter 和 python-jose 來實現這些功能。

性能優化與擴展

使用緩存技術

# config/caching.py
from fastapi import Request
from typing import Any
from fastapi.responses import Response

class CacheInterceptor:
    def __init__(self, cache):
        self.cache = cache

    async def intercept_request(self, request: Request) -> Any:
        key = f"{request.method}_{request.url}"
        value = await self.cache.get(key)
        if value is not None:
            return Response(value)

    async def intercept_response(self, response: Response) -> None:
        key = f"{response.request.method}_{response.request.url}"
        value = str(response.body(), encoding="utf-8")
        await self.cache.set(key, value)

分佈式事務管理

# config/transaction.py
from fastapi import Depends
from typing import Any
import asyncio

class TransactionManager:
    def __init__(self):
        pass

    async def begin(self):
        pass

    async def commit(self):
        await asyncio.sleep(1)  # 示例事務提交時間

    async def rollback(self):
        pass

即時監控與日誌記錄

# config/monitoring.py
import logging
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

Instrumentator.instrument(app)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

項目部署與維護

使用 Docker 部署

創建 Dockerfile：

FROM python:3.8-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "--host", "0.0.0.0", "--port", "8000", "main:app"]

構建鏡像並運行：

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

使用 Kubernetes 擴展

創建 Deployment 和 Service 文件：

# deployment.yml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server-deployment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
    spec:
      containers:
        - name: mcp-server
          image: mcp-server:latest
          ports:
            - containerPort: 8000

# service.yml
apiVersion: v1
kind: Service
metadata:
  name: mcp-server-service
spec:
  type: LoadBalancer
  ports:
    - port: 80
      targetPort: 8000
  selector:
    app: mcp-server

應用配置：

kubectl apply -f deployment.yml -f service.yml

使用 Prometheus 監控

配置 Prometheus 監控 Kubernetes 集群中的 MCP Server：

# prometheus.yml
scrape_configs:
  - job_name: 'mcp-server'
    static_configs:
      - targets: ['<kubernetes-master-ip>:80']

使用 Grafana 可視化

創建 Dashboard 來展示監控數據，集成不同的圖表和警報規則。

🔧 技術細節

核心功能技術細節

• 跨雲平臺兼容性：通過封裝不同雲服務提供商的 API 接口，實現對 AWS、GCP、Azure 等主流雲服務的支持。
• 負載均衡與故障轉移：採用算法根據雲實例的負載和可用性，自動分配任務，同時在某個實例出現故障時，能快速將任務轉移到其他可用實例。
• 日誌與監控集成：內置對 Prometheus 和 Grafana 的支持，通過在代碼中集成相應的 SDK，實現即時數據的採集和展示。
• 安全認證：支持 OAuth2、JWT 等多種身份驗證機制，在 API 接口層進行身份驗證和授權，保障數據安全。

高級功能技術細節

• 分佈式事務管理：集成 atom 或 orchestrator 等工具，確保跨雲服務的事務一致性。
• 緩存與加速：集成 Redis 或 Memcached 進行結果緩存，通過在代碼中添加緩存邏輯，提升響應速度。
• 動態擴展：根據負載自動調整資源使用，通過集成 Kubernetes 或 Elastic Beanstalk，實現計算資源的自動擴展。
• 智能路由：基於地理位置或策略分配請求到最佳可用節點，通過自定義路由邏輯，根據即時指標和預設策略進行請求分發。

📄 許可證

文檔中未提及相關許可證信息。

最佳實踐

1. 模塊化設計：將功能拆分成獨立的模塊或服務，便於維護和擴展。
2. 依賴管理：使用工具如 poetry 或 pipenv 管理 Python 依賴，確保環境一致性。
3. 配置管理：使用配置文件而不是硬編碼值，方便在不同環境中部署。
4. 日誌記錄：集成結構化日誌記錄庫如 logging 或 sentry，便於調試和分析問題。
5. 測試驅動開發：編寫單元測試、集成測試和端到端測試，確保代碼質量。

項目總結

通過以上步驟，您可以構建一個高效可靠的 MCP Server。從模塊化的 API 設計到性能優化和擴展，每個細節都至關重要。使用 Docker 和 Kubernetes 進行容器化部署，並結合 Prometheus 和 Grafana 進行監控和可視化，將使您的服務更加穩定和易於管理。