FAST MCP PROD

FastMCP - 企业级AI代理工具编排平台

FastMCP是一个面向企业AI代理工具的生产就绪编排平台。它通过具有企业级安全、RBAC/ABAC强制执行、确认门、令牌代理、速率限制和不可变审计追踪的统一API，提供安全、可扩展且可审计的外部工具和服务访问。。

🏗️ 架构概览

graph TB
    subgraph "AI Agents"
        A1[Agent 1]
        A2[Agent 2]
        A3[Agent N]
    end
    
    subgraph "FastMCP Core"
        API[FastAPI Router]
        AUTH[JWT Auth + RBAC]
        RATE[Rate Limiter]
        CONF[Confirmation Gates]
        AUDIT[Audit Logger]
        BROKER[Token Broker]
    end
    
    subgraph "Provider Layer"
        DB[Database Provider]
        PD[Pipedrive Provider]
        MCP[MCP Provider]
        ADS[Ads Provider]
    end
    
    subgraph "External Services"
        SQLITE[(SQLite DB)]
        PIPE[Pipedrive API]
        MCP1[MCP Server 1]
        MCP2[MCP Server 2]
        MCPN[MCP Server N]
        GOOGLE[Google Ads]
    end
    
    A1 --> API
    A2 --> API
    A3 --> API
    
    API --> AUTH
    AUTH --> RATE
    RATE --> CONF
    CONF --> BROKER
    BROKER --> DB
    BROKER --> PD
    BROKER --> MCP
    BROKER --> ADS
    
    DB --> SQLITE
    PD --> PIPE
    MCP --> MCP1
    MCP --> MCP2
    MCP --> MCPN
    ADS --> GOOGLE
    
    AUTH --> AUDIT
    CONF --> AUDIT
    BROKER --> AUDIT

🔄 请求流程图

sequenceDiagram
    participant Agent
    participant FastMCP
    participant RBAC
    participant RateLimit
    participant Provider
    participant ExternalAPI
    participant Audit
    
    Agent->>FastMCP: POST /tools/{toolId}/invoke
    FastMCP->>FastMCP: Validate JWT & Extract Claims
    FastMCP->>RBAC: Check Permissions
    RBAC-->>FastMCP: Allow/Deny + Confirmation Required?
    
    alt Access Denied
        FastMCP-->>Agent: 403 Forbidden
    else Allowed
        FastMCP->>RateLimit: Check Rate Limits
        alt Rate Limited
            FastMCP-->>Agent: 429 Too Many Requests
        else Within Limits
            alt Requires Confirmation
                FastMCP->>FastMCP: Generate Action Token
                FastMCP->>Audit: Log Pending Action
                FastMCP-->>Agent: 202 Accepted + Action Token
                Agent->>FastMCP: POST /tools/{toolId}/confirm
                FastMCP->>FastMCP: Validate Action Token
            end
            
            FastMCP->>Provider: Route to Provider
            Provider->>ExternalAPI: Make API Call
            ExternalAPI-->>Provider: Response
            Provider-->>FastMCP: Formatted Response
            FastMCP->>Audit: Log Successful Action
            FastMCP-->>Agent: 200 OK + Result
        end
    end

📁 代码库结构

fastmcp/
├── api/                    # FastAPI routes and dependencies
│   ├── deps.py            # JWT validation, session management
│   ├── routes_catalog.py  # Tool catalog endpoints
│   ├── routes_internal.py # Health checks, metrics
│   ├── routes_manifests.py# Tool manifest registration
│   ├── routes_tools.py    # Tool invocation and confirmation
│   └── utils.py           # Request hashing, error responses
├── core/                  # Core business logic
│   ├── config.py          # Environment configuration
│   ├── rate_limit.py      # Token bucket rate limiting
│   ├── rbac.py            # Role-based access control
│   ├── security.py        # JWT handling, correlation IDs
│   └── token_broker.py    # Provider token exchange
├── db/                    # Database layer
│   ├── schemas_sqlmodel.py# SQLModel schemas
│   └── session.py         # Database session management
├── models/                # Pydantic models
│   ├── audit.py           # Audit event models
│   ├── tool_manifest.py   # Tool definition schemas
│   └── user_agent.py      # User agent parsing
├── providers/             # External service integrations
│   ├── base.py            # Provider interface
│   ├── ads_provider.py    # Google Ads integration
│   ├── database_provider.py# SQLite database access
│   ├── mcp_provider.py    # MCP server integration
│   └── pipedrive_provider.py# Pipedrive CRM integration
├── policies/              # RBAC policy definitions
│   └── policies.yml       # Role and rule definitions
├── tests/                 # Test suite
│   ├── conftest.py        # Test fixtures
│   ├── test_happy_path.py # End-to-end success flows
│   ├── test_policy_denial.py# RBAC failure scenarios
│   ├── test_confirmation_flow.py# Confirmation workflows
│   └── test_idempotency.py# Idempotency key handling
├── utils/                 # Utilities
│   ├── audit_logger.py    # Immutable audit logging
│   └── openapi_ingestor.py# OpenAPI spec processing
├── main.py                # FastAPI application entry point
└── pyproject.toml         # Project dependencies

🔐 安全架构

JWT 认证流程

graph LR
    subgraph "Token Validation"
        JWT[JWT Token] --> VERIFY[Verify Signature]
        VERIFY --> CLAIMS[Extract Claims]
        CLAIMS --> VALIDATE[Validate Expiry]
    end
    
    subgraph "Claims Structure"
        SUB[Subject: agent-id]
        TENANT[Tenant: organization]
        ROLES[Roles: [finance-write]]
        SCOPES[Scopes: [ads.write]]
    end
    
    VALIDATE --> SUB
    VALIDATE --> TENANT
    VALIDATE --> ROLES
    VALIDATE --> SCOPES

基于角色的访问控制（RBAC）策略引擎

# policies/policies.yml
roles:
  agent: ["tools.invoke:read", "tools.query:catalog"]
  finance-write: ["ads.write", "tools.invoke:write"]
  sales-write: ["pipedrive.write", "tools.invoke:write"]
  data-analyst: ["database.read", "database.execute", "tools.invoke:write"]
  data-admin: ["database.read", "database.execute", "database.write", "tools.invoke:write"]
  mcp-user: ["mcp.filesystem", "mcp.database", "mcp.git", "tools.invoke:write"]
  mcp-admin: ["mcp.*", "tools.invoke:write"]

rules:
  - id: invoke-ads-create
    effect: allow
    when:
      action: tools.invoke
      resource.toolId: "ads:createCampaign"
      subject.roles: ["finance-write"]
    require_confirmation: true

🛠️ 服务提供商系统

提供者接口

所有提供商均实施了 ProviderAdapter 基类：

class ProviderAdapter:
    def __init__(self, provider_id: str):
        self.provider_id = provider_id
    
    async def exchange(self, scopes: list[str], subject: str, tenant: str, purpose: str) -> dict:
        """Exchange FastMCP token for provider-specific credentials"""
        pass
    
    async def call(self, endpoint: str, payload: dict) -> dict:
        """Execute tool call on external service"""
        pass

可用的提供商

1. 数据库提供者（database_provider.py)

• 目的安全执行SQL查询
• 工具： listTables， getSchema， executeQuery
• 安全查询验证、参数化执行、行数限制
• 异步用途 asyncio.to_thread 用于非阻塞的SQLite访问

2. Pipedrive 提供者（或供应商）pipedrive_provider.py)

• 目的客户关系管理（CRM）运营
• 工具: createDeal， getDeal， updateDeal， createContact
• 认证（或授权）基于API令牌的身份验证
• 配置： PIPEDRIVE_API_TOKEN， PIPEDRIVE_COMPANY_DOMAIN

3. MCP 提供商（mcp_provider.py）

• 目的与模型上下文协议服务器的集成
• 特点/功能自动发现，带退避的重试，动态工具注册
• 配置环境变量 MCP_*_URL
• 可扩展性通过添加环境变量来添加新的MCP服务器

4. 广告提供商（ads_provider.py）

• 目的Google Ads广告系列管理
• 工具： createCampaign
• 安全需要对财务操作进行确认

🚀 如何添加组件

添加新代理

1. 生成JWT令牌 具有适当的角色和范围：

import time
from jose import jwt
from fastmcp.core.config import get_dev_crypto_material

_, private_key = get_dev_crypto_material()
now = int(time.time())

payload = {
    "iss": "https://idp.local",
    "aud": "fastmcp",
    "sub": "my-agent-id",
    "tenant": "my-organization",
    "roles": ["data-analyst", "mcp-user"],  # Define agent capabilities
    "scopes": ["database.read", "mcp.filesystem", "tools.invoke:write"],
    "iat": now,
    "exp": now + 3600
}

token = jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": "dev-key"})

1. 在请求中使用令牌：

curl -H "Authorization: Bearer $TOKEN" http://localhost:8001/tool-catalog

定义代理范围

编辑 fastmcp/policies/policies.yml：

roles:
  my-custom-role: ["custom.scope", "tools.invoke:write"]

rules:
  - id: custom-tool-access
    effect: allow
    when:
      action: tools.invoke
      resource.provider_id: "my-provider"
      subject.roles: ["my-custom-role"]

添加新工具

1. 创建提供者 （如需）：

# fastmcp/providers/my_provider.py
from .base import ProviderAdapter, register_adapter

class MyProviderAdapter(ProviderAdapter):
    def __init__(self):
        super().__init__("my-provider")
    
    async def exchange(self, scopes, subject, tenant, purpose):
        return {"access_token": "my-token", "token_type": "Bearer"}
    
    async def call(self, endpoint: str, payload: dict) -> dict:
        if endpoint == "myTool":
            return await self.my_tool(payload)
        raise ValueError(f"Unknown endpoint: {endpoint}")
    
    async def my_tool(self, payload):
        # Implement tool logic
        return {"result": "success"}

register_adapter(MyProviderAdapter())

1. 注册提供商 在 main.py:

from fastmcp.providers import my_provider  # noqa: F401

1. 创建工具清单：

MY_TOOL_MANIFEST = {
    "toolId": "my-provider:myTool",
    "name": "My Custom Tool",
    "description": "Does something useful",
    "inputs": {
        "type": "object",
        "properties": {
            "param1": {"type": "string", "description": "First parameter"}
        },
        "required": ["param1"]
    },
    "outputs": {"type": "object"},
    "required_scopes": ["my-provider.use"],
    "provider_id": "my-provider",
    "tenant": "public"
}

1. 添加到启动播种 在 main.py:

def seed_database():
    # ... existing code ...
    
    my_manifest = ToolManifest(**MY_TOOL_MANIFEST)
    with Session(engine) as session:
        existing = session.get(Manifest, my_manifest.toolId)
        if not existing:
            session.add(Manifest(
                toolId=my_manifest.toolId,
                manifest=my_manifest.model_dump(),
                tenant=my_manifest.tenant,
                provider_id=my_manifest.provider_id,
            ))
        session.commit()

添加新的MCP服务器

1. 启动您的MCP服务器 在某个端口上（例如，3004）

1. 添加到环境 (.env)：

MCP_MYSERVER_URL=http://localhost:3004

1. 重启FastMCP - 工具是自动发现的！

1. 验证发现：

python test_mcp.py

MCP 提供商将自动：

• 通过JSON-RPC发现工具 tools/list
• 生成FastMCP清单
• 在数据库中注册工具
• 将路由调用转发到MCP服务器

🔄 工作流和回退机制

工具调用工作流

flowchart TD
    START([Agent Request]) --> VALIDATE{Valid JWT?}
    VALIDATE -->|No| AUTH_ERROR[401 Unauthorized]
    VALIDATE -->|Yes| RBAC{RBAC Check}
    
    RBAC -->|Denied| FORBIDDEN[403 Forbidden]
    RBAC -->|Allowed| RATE{Rate Limit?}
    
    RATE -->|Exceeded| RATE_ERROR[429 Too Many Requests]
    RATE -->|OK| CONFIRM{Requires Confirmation?}
    
    CONFIRM -->|Yes| PENDING[202 + Action Token]
    CONFIRM -->|No| EXECUTE[Execute Tool]
    
    PENDING --> CONFIRM_REQ[Agent Confirms]
    CONFIRM_REQ --> VALIDATE_TOKEN{Valid Token?}
    VALIDATE_TOKEN -->|No| INVALID[401 Invalid Token]
    VALIDATE_TOKEN -->|Yes| EXECUTE
    
    EXECUTE --> PROVIDER[Route to Provider]
    PROVIDER --> EXTERNAL[External API Call]
    EXTERNAL --> SUCCESS[200 + Result]
    EXTERNAL --> PROVIDER_ERROR[502 Provider Error]
    
    SUCCESS --> AUDIT[Log Audit Event]
    PROVIDER_ERROR --> AUDIT
    AUTH_ERROR --> AUDIT
    FORBIDDEN --> AUDIT

回退机制

1. 提供者故障

• 超时处理对外部调用设置30秒超时
• 错误包装提供者异常变为502响应
• 审计日志记录所有故障均记录有相关联的ID

2. 数据库故障

• 连接重试SQLite在连接丢失时会自动重新连接
• 事务回滚失败的操作不会破坏状态
• 优雅降级如果数据库（DB）故障，服务继续运行，不进行审计

3. MCP服务器发现

• 使用退避策略重试3次尝试，采用指数退避策略（500毫秒、1秒、2秒）
• 非致命性故障如果MCP服务器宕机，启动过程继续进行
• 部分发现可用服务器已注册，不可用服务器已跳过

4. 限速（或速率限制）

• 内存回退（或内存降级）如果 Redis 不可用，则使用内存中的存储桶
• 优雅降级如果两者都失败，则禁用速率限制
• 租户级隔离一个租户的故障不会影响其他租户

5. JWT 验证

• 密钥轮换支持JWKS中的多个密钥
• 时钟偏差容限对于exp/iat索赔的5分钟宽限期
• 优雅的错误处理无效令牌返回401，而非500

🧪 测试

运行测试

# Install test dependencies
pip install pytest anyio

# Run test suite
PYTHONPATH=. python -m pytest fastmcp/tests/ -v

# Run with coverage
PYTHONPATH=. python -m pytest fastmcp/tests/ --cov=fastmcp

测试类别

1. “Happy Path”翻译成中文是“顺利路径”或“理想路径” （test_happy_path.py)

- 完整目录 → 调用 → 确认流程 - 审计轨迹验证 - 代币元数据验证

1. 政策否认 (test_policy_denial.py)

- 基于角色的访问控制（RBAC）实施 - 范围处理不足 - 租户隔离

1. 确认流程 (test_confirmation_flow.py)

- 动作令牌生成 - 令牌过期处理 - 有效载荷验证

1. 幂等性 (test_idempotency.py)

- 处理重复请求 - 缓存重放行为 - TTL（生存时间）到期

手动测试脚本

数据库测试

python test_database.py

测试具有不同角色和查询类型的数据库提供者。

MCP测试

python test_mcp.py

测试MCP服务器发现和工具调用。

🔧 配置

环境变量

# Core Configuration
APP_NAME=fastmcp
JWT_ISSUER=https://idp.local
JWT_AUDIENCE=fastmcp
JWT_ALG=RS256
JWT_TTL_MIN=10

# Security
REQUIRE_MTLS=false
CONFIRM_TTL_SEC=300
ACTION_TOKEN_BYTES=24

# Rate Limiting
RATE_LIMIT_DEFAULT_RPS=5
RATE_BUCKET_BURST=10
REDIS_URL=redis://localhost:6379  # Optional

# Database
SQLITE_URL=sqlite:///./fastmcp.db
DATABASE_PATH=./data/app.db
DATABASE_MAX_ROWS=1000

# Audit
AUDIT_WORM_DIR=.audit

# Provider Configurations
PIPEDRIVE_API_TOKEN=your_token_here
PIPEDRIVE_COMPANY_DOMAIN=your_company

# MCP Servers (auto-discovered)
MCP_FILESYSTEM_URL=http://localhost:3001
MCP_DATABASE_URL=http://localhost:3002
MCP_GIT_URL=http://localhost:3003
MCP_DISCOVERY_RETRIES=3
MCP_DISCOVERY_BACKOFF_MS=500

策略配置

RBAC系统通过以下方式配置： fastmcp/policies/policies.yml：

version: 1

# Role to scope mapping
roles:
  agent: ["tools.invoke:read", "tools.query:catalog"]
  finance-write: ["ads.write", "tools.invoke:write"]
  sales-write: ["pipedrive.write", "tools.invoke:write"]
  data-analyst: ["database.read", "database.execute", "tools.invoke:write"]
  data-admin: ["database.read", "database.execute", "database.write", "tools.invoke:write"]
  mcp-user: ["mcp.filesystem", "mcp.database", "mcp.git", "tools.invoke:write"]
  mcp-admin: ["mcp.*", "tools.invoke:write"]

# Access control rules
rules:
  - id: catalog-visibility
    effect: allow
    when:
      action: catalog:list

  - id: invoke-ads-create
    effect: allow
    when:
      action: tools.invoke
      resource.toolId: "ads:createCampaign"
      subject.roles: ["finance-write"]
    require_confirmation: true

  - id: invoke-database-tools
    effect: allow
    when:
      action: tools.invoke
      resource.provider_id: "database"
      subject.roles: ["data-analyst", "data-admin"]

  - id: invoke-mcp-tools
    effect: allow
    when:
      action: tools.invoke
      resource.provider_id: "mcp"
      subject.roles: ["mcp-user", "mcp-admin"]

  - id: safety-destroy-requires-confirm
    effect: allow
    when:
      action: tools.invoke
      resource.safety_tags_any: ["financial", "destructive"]
    require_confirmation: true

  - id: tenant-isolation
    effect: deny
    when:
      resource.tenant: "!= subject.tenant"

📊 监控与可观测性

审计日志记录

• 位置: .audit/audit-YYYYMMDD.jsonl
• 格式带有关联ID的结构化JSON
• 不可变性哈希链表条目防止篡改
• 留存每日轮换，可配置保留策略

结构化日志记录

# All logs include correlation IDs and structured data
logger.info("tool.invoked", 
           tool_id="database:executeQuery",
           subject="agent-1",
           tenant="public",
           correlation_id="abc123")

健康检查

curl http://localhost:8001/health
# Returns: {"status": "healthy", "timestamp": "..."}

🚀 部署

生产检查清单

1. 安全

- \[ \] 将开发环境的JWKS替换为生产环境的密钥 - \[ \] 设置 REQUIRE_MTLS=true 用于生产 - \[ \] 使用正确的JWT签发者和受众 - \[ \] 保护提供商API令牌

1. 可扩展性

- \[ \] 配置Redis以实现速率限制 - \[ \] 设置适当的数据库（推荐使用PostgreSQL） - \[ \] 配置日志聚合 - \[ \] 设置监控和警报

1. 高可用性

- \[ \] 部署多个FastMCP实例 - \[ \] 使用带有健康检查的负载均衡器 - \[ \] 配置数据库复制 - \[ \] 设置备份和恢复

Docker 部署

FROM python:3.12-slim

WORKDIR /app
COPY fastmcp/ ./fastmcp/
COPY requirements.txt .

RUN pip install -r requirements.txt

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

🤝 贡献/参与

开发环境设置

# Clone repository
git clone https://github.com/Yashkalwar/FAST_MCP_PROD.git
cd FAST_MCP_PROD

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # Linux/Mac
# or
.venv\Scripts\activate     # Windows

# Install dependencies
pip install -r requirements.txt

# Run tests
pytest fastmcp/tests/

# Start development server
uvicorn fastmcp.main:app --reload --port 8001

代码风格

• 遵循PEP 8规范
• 使用类型提示
• 为公共API添加文档字符串
• 包含对新功能的测试

拉取请求流程

1. 克隆该仓库
2. 创建一个特性分支
3. 为新功能添加测试
4. 确保所有测试通过
5. 更新文档
6. 提交拉取请求

📚 API 参考

认证

所有请求都需要在Authorization头中包含一个有效的JWT令牌：

Authorization: Bearer 

终点（或终点参数）

获取/工具目录

列出已认证代理可用的工具。

回应：

{
  "tools": [
    {
      "toolId": "database:listTables",
      "name": "List Database Tables",
      "description": "List all tables in the database",
      "inputs": {"type": "object"},
      "outputs": {"type": "object"},
      "required_scopes": ["database.read"],
      "provider_id": "database"
    }
  ]
}

POST /tools/{toolId}/invoke 翻译为中文是：“向/tools/{toolId}/invoke发送POST请求”

使用给定的参数调用一个工具。

请求：

{
  "param1": "value1",
  "param2": "value2"
}

响应（成功）：

{
  "toolId": "database:listTables",
  "result": {
    "tables": [{"name": "users", "row_count": 100}]
  },
  "token_meta": {
    "token_type": "Bearer",
    "expires_in": 3600
  }
}

响应（需确认）：

{
  "requires_confirmation": true,
  "action_token": "abc123...",
  "expires_at": "2025-01-01T12:00:00Z"
}

POST /tools/{toolId}/confirm 翻译为中文是：“发送到 /tools/{toolId}/confirm（确认工具）” 或者更简洁地表达为：“确认工具（POST 请求）”。这里，“POST”表示请求方法，“/tools/{toolId}/confirm”是请求的路径，其中{toolId}是一个占位符，代表具体的工具ID，“confirm”表示确认操作

确认需要审批的工具调用。

请求：

{
  "action_token": "abc123...",
  "param1": "value1",
  "param2": "value2"
}

POST /manifests/register 翻译为中文是：“提交/清单/注册”。不过，这里的“POST”是HTTP方法，通常在实际应用中不会直接翻译，而是保持原样以表示请求方法。所以，更常见的表达可能是“发送 POST 请求到 /manifests/register 路径以进行注册”。但简化的翻译就是“提交/清单/注册”

注册一个新的工具清单（需要管理员角色）。

请求：

{
  "toolId": "my-provider:myTool",
  "name": "My Tool",
  "description": "Does something useful",
  "inputs": {"type": "object"},
  "outputs": {"type": "object"},
  "required_scopes": ["my-provider.use"],
  "provider_id": "my-provider"
}

🔍 故障排除

常见问题

1. “未找到工具清单”（404）

• 原因工具未注册或代理无访问权限
• 解决方案检查工具目录，验证RBAC策略

2. “访问被拒绝”（403）

• 原因权限不足
• 解决方案检查JWT角色/范围，更新策略

3. “请求速率过高”（429）

• 原因请求过多
• 解决方案实现退避机制，检查速率限制配置

4. “服务提供者调用失败”（502）

• 原因外部服务不可用
• 解决方案检查提供商配置，网络连接情况

5. 未发现MCP工具

• 原因MCP服务器未运行或配置错误
• 解决方案检查MCP服务器URL，重启FastMCP

调试模式

# Enable debug logging
export LOG_LEVEL=DEBUG
uvicorn fastmcp.main:app --reload --log-level debug

审计轨迹分析

# View recent audit events
tail -f .audit/audit-$(date +%Y%m%d).jsonl | jq

# Search for specific tool invocations
grep "database:executeQuery" .audit/audit-*.jsonl | jq

📄 许可证

此项目采用MIT许可证授权。详见LICENSE文件。

🆘 支持

• 问题：
• 文档这个README文件和内联代码注释
• 社区: 讨论

______________________________________________________________________

FastMCP - 安全、可扩展且可审计的AI代理工具编排。为企业而建，为开发者而设计。