AI代理上下文管理服务器

MCP Server是一个持久化后端服务,通过集中管理内存、工具和逻辑,减少AI代理上下文窗口膨胀问题。

  • by GeoAziz
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 GeoAziz/mcp-server

使用说明

MCP服务器-模型上下文协议

⚠️ 架构更新:单片 mcp_server.py 现在已弃用,取而代之的是模块化 app/main.py 结构。两者在功能上完全相同,但新代码应使用模块化版本。 迁移指南

一种持久的后端服务,通过集中内存、工具和逻辑来减少AI代理上下文窗口的膨胀。

🎯 它解决的问题

之前(工具提示):

  • 每次LLM调用中都会重复工具模式
  • 内存/状态不断重新发送
  • 上下文窗口填充速度很快
  • 多代理支持不佳

之后(MCP服务器):

  • 工具在外部服务器上运行
  • 内存在会话中持续存在
  • 精益提示(代币减少10倍)
  • 可扩展的多代理架构

🚀 快速开始

1.安装依赖项

pip install -r requirements.txt

2.启动服务器

生产(推荐):

python -m uvicorn app.main:app --host 0.0.0.0 --port 8000

自动重新加载开发:

python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

服务器运行时间: http://localhost:8000

可选环境变量:

# Enable API key authentication (recommended for production)
export MCP_API_KEY="your-secret-key"

# GitHub API token (optional but recommended for higher rate limits)
export MCP_GITHUB_TOKEN="your-github-token"

# Figma API token (required for Figma actions)
export MCP_FIGMA_TOKEN="your-figma-token"

# Configure CORS allowed origins (default: *)
export MCP_CORS_ORIGINS="https://yourdomain.com,https://app.yourdomain.com"

# Configure rate limit (default: 100/minute)
export MCP_RATE_LIMIT="200/minute"

# Configure log retention limit (default: 1000)
export MCP_LOG_RETENTION="5000"

3.与客户进行测试

python mcp_client_example.py

📖 API文档

交互式API文档

MCP服务器提供由OpenAPI(Swagger)和ReDoc提供支持的自动生成交互式API文档:

Swagger UI(交互式):

http://localhost:8000/docs
  • 直接从浏览器试用API端点
  • 查看请求/响应示例
  • 请参阅详细的参数说明

ReDoc(干净文档):

http://localhost:8000/redoc
  • 干净、可搜索的API参考
  • 三面板设计,便于导航
  • 移动友好界面

OpenAPI规范(JSON):

http://localhost:8000/openapi.json
  • 下载完整的OpenAPI 3.1.0规范
  • 与代码生成器(OpenAPI生成器、Swagger Codegen)一起使用
  • 导入API测试工具(Postman、Insomnia)

API快速参考

所有端点都以这种标准格式返回响应:

{
  "success": true,
  "data": { ... },
  "message": "Action completed successfully",
  "timestamp": "2026-02-16T16:00:00.000000"
}

响应字段:

  • success (boolean)-操作是否成功
  • data (任意)-响应数据(因端点而异)
  • message (string)-人类可读状态消息
  • timestamp (字符串)-ISO 8601时间戳

📚 API概述

API版本

MCP服务器支持API版本控制,以确保向后兼容性,并允许在不破坏现有客户端的情况下进行未来的增强。

当前API版本:

  • 第1版 (/api/v1/*)-当前稳定版本(推荐)
  • 第2版 (/api/v2/*)-用于未来扩展的占位符
  • 遗产 (/mcp/*)-原始端点(已弃用,为向后兼容而保留)

迁移指南: 所有遗产 /mcp/* 端点位于 /api/v1/*例如:

  • /mcp/state → /api/v1/state
  • /mcp/query → /api/v1/query
  • /mcp/logs → /api/v1/logs
  • /mcp/reset → /api/v1/reset

核心终点

V1端点(推荐)

端点 方法 描述
/ GET 运行状况检查(显示可用的API版本)
/api/v1/query POST 主要操作端点
/api/v1/state GET 获取内存快照(支持过滤和分页)
/api/v1/logs GET 查看具有可配置保留期的结构化操作日志
/api/v1/reset POST 重置所有内存

传统端点(已弃用)

端点 方法 描述 替换
/mcp/query POST 主操作端点 使用 /api/v1/query
/mcp/state GET 获取内存快照 使用 /api/v1/state
/mcp/logs GET 查看操作日志 使用 /api/v1/logs
/mcp/reset POST 重置所有内存 使用 /api/v1/reset

注: 为了向后兼容,保留了传统端点,但可能会在未来的主要版本中删除。请迁移到版本化端点。

可用操作

用户管理:

  • list_users -获取所有用户
  • add_user -添加新用户
  • remove_user -删除用户
  • get_user -获取用户详细信息

任务管理:

  • list_tasks -列出所有任务(带过滤器)
  • add_task -创建新任务
  • update_task -更新任务
  • delete_task -删除任务
  • search_tasks -按查询搜索

配置:

  • get_config -获取配置值
  • update_config -更新配置

公用设施:

  • calculate -执行计算
  • summarize_data -获取数据摘要

github:

  • github_search_repositories -搜索仓库
  • github_search_issues -搜索问题/PR
  • github_get_repository -回购详情
  • github_list_issues -列出问题
  • github_list_pulls -列出PR

Figma:

  • figma_get_file -文件元数据
  • figma_get_nodes -节点元数据
  • figma_get_components -文件组件
  • figma_get_styles -文件样式

剧作家:

  • playwright_get_title -页面标题
  • playwright_get_text -页面文本
  • playwright_screenshot -屏幕截图

💡 使用示例

Python客户端

from mcp_client_example import MCPClient

client = MCPClient()

# Add user
client.add_user("alice")

# Create task
task = client.add_task(
    title="Build feature X",
    priority="high",
    assigned_to="alice"
)

# List tasks for user
alice_tasks = client.list_tasks(assigned_to="alice")

# Get summary
summary = client.get_summary()

直接HTTP(curl)

使用V1 API(推荐):

用户管理

# List all users
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "list_users",
    "params": {}
  }'

# Add a new user
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "add_user",
    "params": {
      "username": "alice",
      "role": "admin",
      "metadata": {
        "team": "engineering",
        "location": "San Francisco"
      }
    }
  }'

# Get user details
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "get_user",
    "params": {
      "username": "alice"
    }
  }'

# Remove a user
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "remove_user",
    "params": {
      "username": "alice"
    }
  }'

任务管理

# Create a new task
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "add_task",
    "params": {
      "title": "Implement API documentation",
      "description": "Add comprehensive OpenAPI/Swagger docs",
      "priority": "high",
      "assigned_to": "alice"
    }
  }'

# List all tasks
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "list_tasks",
    "params": {}
  }'

# List tasks for a specific user
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "list_tasks",
    "params": {
      "assigned_to": "alice"
    }
  }'

# List tasks by status
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "list_tasks",
    "params": {
      "status": "pending"
    }
  }'

# Update a task
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "update_task",
    "params": {
      "task_id": 1,
      "status": "completed",
      "priority": "medium"
    }
  }'

# Delete a task
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "delete_task",
    "params": {
      "task_id": 1
    }
  }'

# Search tasks by keyword
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "search_tasks",
    "params": {
      "query": "documentation"
    }
  }'

配置管理

# Get all configuration
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "get_config",
    "params": {}
  }'

# Get specific config value
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "get_config",
    "params": {
      "key": "app_name"
    }
  }'

# Update configuration
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "update_config",
    "params": {
      "key": "app_name",
      "value": "My MCP Server"
    }
  }'

公用事业行动

# Perform calculations
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "calculate",
    "params": {
      "operation": "sum",
      "numbers": [10, 20, 30, 40]
    }
  }'

# Get data summary
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -d '{
    "action": "summarize_data",
    "params": {}
  }'

状态和日志

# Get full state
curl http://localhost:8000/api/v1/state

# Get filtered state (tasks only, first 10)
curl "http://localhost:8000/api/v1/state?entity=tasks&limit=10"

# Get pending tasks
curl "http://localhost:8000/api/v1/state?entity=tasks&status=pending"

# Get logs
curl http://localhost:8000/api/v1/logs?limit=10

# Reset all memory (use with caution!)
curl -X POST http://localhost:8000/api/v1/reset

通过身份验证

当 MCP_API_KEY 环境变量已设置,请在所有请求中包含API键:

# Using API key authentication
curl -X POST http://localhost:8000/api/v1/query \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-key-here" \
  -d '{
    "action": "list_users",
    "params": {}
  }'

# Get state with authentication
curl http://localhost:8000/api/v1/state \
  -H "X-API-Key: your-secret-key-here"

遗留终结点(仍然有效,但已弃用):

# Legacy endpoints - replace /mcp/ with /api/v1/
curl http://localhost:8000/mcp/state  # Deprecated
curl http://localhost:8000/api/v1/state  # Use this instead

状态端点查询参数

这 /api/v1/state (和遗产 /mcp/state)端点支持用于筛选和分页的可选查询参数:

  • entity:按实体类型筛选(users | tasks | config | logs)
  • limit:要退回的最大项目数
  • offset:要跳过的项目数(用于分页)
  • status:按状态筛选任务(仅适用于以下情况 entity=tasks)

示例:

# Get only tasks
curl http://localhost:8000/api/v1/state?entity=tasks

# Get first 5 pending tasks
curl http://localhost:8000/api/v1/state?entity=tasks&status=pending&limit=5

# Get users with pagination (skip first 10, return next 20)
curl http://localhost:8000/api/v1/state?entity=users&offset=10&limit=20

来自AI代理提示

与其在LLM提示符中定义工具,不如给它一个端点:

# Old way (bloated prompt):
prompt = """
You have access to these tools:
[...massive tool schemas...]
[...memory state...]
"""

# New way (lean prompt):
prompt = """
You can query the MCP server at http://localhost:8000/api/v1/query

Available actions: list_users, add_task, list_tasks, etc.

Example:
POST /api/v1/query
{
  "action": "list_users",
  "params": {}
}
"""

🔧 VS代码集成

进行完整的端到端演练(启动服务器→ IDE → 工具调用),请参阅 docs/IDE_INTEGRATION.md. 有关集成操作目录,请参见 docs/MCP_INTEGRATIONS.md.

选项1:与Copilot/GitHub Copilot聊天一起使用

您的AI助手可以直接从生成的代码调用MCP服务器:

import requests

def get_user_tasks(username):
    response = requests.post(
        "http://localhost:8000/api/v1/query",
        json={
            "action": "list_tasks",
            "params": {"assigned_to": username}
        }
    )
    return response.json()["data"]

选项2:添加为VS代码任务

创建 .vscode/tasks.json:

{
  "version": "2.0.0",
  "tasks": [
    {
      "label": "Start MCP Server",
      "type": "shell",
      "command": "python",
      "args": ["mcp_cli.py", "start"],
      "isBackground": true,
      "problemMatcher": []
    }
  ]
}

运行方式: Terminal > Run Task > Start MCP Server

选项3:调试配置

创建 .vscode/launch.json:

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "MCP Server",
      "type": "python",
      "request": "launch",
      "program": "${workspaceFolder}/mcp_cli.py",
      "args": ["start", "--reload"],
      "console": "integratedTerminal"
    }
  ]
}

使用断点进行调试: F5

📊 监控和日志

结构化日志记录

所有操作均以结构化JSON格式记录,包括:

  • 时间戳:ISO 8601操作时间戳
  • 行动:执行的操作
  • 有效载荷:操作参数和结果(截断为200个字符)
  • 状态: success 或 error

查看最近的操作:

# Using v1 API (recommended)
curl http://localhost:8000/api/v1/logs?limit=10

# Legacy endpoint (deprecated)
curl http://localhost:8000/mcp/logs?limit=10

日志条目示例:

{
  "timestamp": "2026-02-16T15:14:57.661938",
  "action": "add_user",
  "payload": {
    "params": {"username": "alice"},
    "result": "{'username': 'alice', 'added': True}"
  },
  "status": "success"
}

日志保留

日志会自动修剪以保持配置的保留限制(默认值:1000个条目)。 通过环境变量配置:

export MCP_LOG_RETENTION="5000"  # Keep last 5000 log entries

访问日志

经由 /api/v1/logs 端点(推荐):

# Get last 10 logs
curl http://localhost:8000/api/v1/logs?limit=10

经由 /api/v1/state 带有实体筛选器的端点:

# Get logs with pagination
curl "http://localhost:8000/api/v1/state?entity=logs&limit=20&offset=10"

传统端点 (已弃用):

curl http://localhost:8000/mcp/logs?limit=10
curl "http://localhost:8000/mcp/state?entity=logs&limit=20&offset=10"

检查服务器运行状况:

curl http://localhost:8000/

🔄 添加自定义工具

首选(模块化应用程序):添加服务并在v1路由器中注册。

# app/services/notification_service.py
async def send_notification(params: Dict[str, Any], db: Session) -> Dict[str, Any]:
  user = params.get("user")
  message = params.get("message")
  if not user or not message:
    raise ValueError("user and message are required")
  return {"sent": True, "user": user}

# app/routers/v1.py
handlers = {
  # ... existing handlers ...
  "send_notification": notification_service.send_notification,
}

Legacy(单片应用程序):添加新操作 mcp_server.py:

# 1. Add handler function
async def handle_my_custom_tool(params: Dict[str, Any]) -> Any:
    # Your logic here
    result = params.get("input") * 2
    return {"result": result}

# 2. Register in handlers dict
handlers = {
    # ... existing handlers ...
    "my_custom_tool": handle_my_custom_tool,
}

然后称之为:

client.query("my_custom_tool", {"input": 42})

🗄️ 数据库集成

对于生产,用数据库替换内存存储:

SQLite示例

import sqlite3

class DatabaseMemory:
    def __init__(self, db_path="mcp.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.init_tables()
    
    def init_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY,
                title TEXT,
                priority TEXT,
                assigned_to TEXT,
                created_at TEXT
            )
        """)
        self.conn.commit()
    
    def add_task(self, task):
        self.conn.execute(
            "INSERT INTO tasks (title, priority, assigned_to, created_at) VALUES (?, ?, ?, ?)",
            (task["title"], task["priority"], task["assigned_to"], task["created_at"])
        )
        self.conn.commit()

PostgreSQL示例

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="mcp_db",
    user="user",
    password="password"
)

🚀 生产部署

码头工人

创建 Dockerfile:

FROM python:3.11-slim

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

COPY . .

EXPOSE 8000
CMD ["python", "mcp_cli.py", "start", "--host", "0.0.0.0", "--port", "8000"]

构建并运行:

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

系统(Linux)

创建 /etc/systemd/system/mcp-server.service:

[Unit]
Description=MCP Server
After=network.target

[Service]
Type=simple
User=youruser
WorkingDirectory=/path/to/mcp
ExecStart=/usr/bin/python3 /path/to/mcp/mcp_cli.py start
Restart=always

[Install]
WantedBy=multi-user.target

启用:

sudo systemctl enable mcp-server
sudo systemctl start mcp-server

🔐 安全

API密钥验证

API密钥身份验证是内置的,可以通过设置环境变量来启用:

export MCP_API_KEY="your-secret-key-here"
python mcp_cli.py start

然后在请求中包含API密钥:

# Using v1 API (recommended)
curl http://localhost:8000/api/v1/state \
  -H "X-API-Key: your-secret-key-here"

# Legacy endpoint
curl http://localhost:8000/mcp/state \
  -H "X-API-Key: your-secret-key-here"

如果 MCP_API_KEY 未设置,身份验证已禁用(对开发有用)。

CORS配置

默认情况下,CORS是启用的,用于开发的通配符来源。对于生产,仅限于特定领域:

# Allow specific origins (comma-separated)
export MCP_CORS_ORIGINS="https://yourdomain.com,https://app.yourdomain.com"
python mcp_cli.py start

CORS中间件支持:

  • 可配置的允许来源
  • 凭据支持
  • 标准HTTP方法(GET、POST、PUT、DELETE、OPTIONS)
  • 允许所有标题

速率限制

自动启用速率限制以防止滥用:

# Configure rate limit (default: 100/minute)
export MCP_RATE_LIMIT="200/minute"
python mcp_cli.py start

支持的格式:

  • 100/minute -每分钟100个请求
  • 10/second -每秒10个请求
  • 1000/hour -每小时1000个请求

当超过速率限制时,API返回:

  • 状态: 429 Too Many Requests
  • 答复: {"error":"Rate limit exceeded: 100 per 1 minute"}

速率限制适用于所有端点,并针对每个IP地址强制执行。

📈 性能提示

  1. 使用连接池 用于数据库连接
  2. 添加缓存 用于频繁访问的数据(Redis)
  3. 配置速率限制 根据您的流量模式(请参阅安全部分)
  4. 使用异步处理程序 用于I/O操作
  5. 添加分页 对于大型结果集
  6. 设置适当的CORS源 减少未经授权的请求

🧪 测试

该项目包括一个全面的基于pytest的测试套件,涵盖所有端点和操作。

运行测试

# Install test dependencies
pip install -r requirements.txt

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run specific test file
pytest tests/test_users.py

# Run with coverage report
pytest --cov=. --cov-report=html

测试覆盖率

该测试套件包括78+个测试,涵盖:

  • 端点:健康检查、状态、查询、日志、重置
  • 用户运营:列出、添加、删除、获取用户详细信息
  • 任务操作:CRUD、搜索、过滤
  • 配置操作:获取并更新配置
  • 认证:API密钥验证和安全
  • 错误处理:输入无效、参数缺失、边缘情况

测试结构

tests/
├── conftest.py              # Pytest fixtures and configuration
├── test_endpoints.py        # Core endpoint tests
├── test_users.py            # User management tests
├── test_tasks.py            # Task management tests
├── test_config.py           # Configuration tests
├── test_auth.py             # Authentication tests
└── test_error_handling.py   # Error handling and edge cases

所有测试都使用内存中的SQLite数据库来提高速度和隔离度。看 tests/README.md 详细文档。

� 从传统单片结构迁移

发生了什么变化

为了清晰和可维护性,代码库已经过重组:

旧位置 新位置 状态
mcp_server.py app/main.py ⚠️ 已弃用
routers/v1.py app/routers/v1.py ⚠️ 已弃用
routers/v2.py app/routers/v2.py ⚠️ 已弃用
models.py app/models/ ⚠️ 已弃用
auth.py app/auth.py ⚠️ 已弃用
database.py app/database.py ⚠️ 已弃用
log_manager.py app/log_manager.py ⚠️ 已弃用

更新什么

旧的启动命令:

python mcp_server.py

新的启动命令(推荐):

python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

API变更

  • 所有端点保持一致
  • /mcp/* 路线现在已标记 @deprecated 但仍能正常工作
  • 更喜欢 /api/v1/* 新客户路线
  • /api/v2/* 现在可用于未来的增强功能

错误处理标准化

  • 全部 ValueError 异常现在会自动返回HTTP 400错误请求
  • 无需更改API用户界面
  • 提高所有端点的一致性

移民的好处

  • ✅ 更清晰的代码组织(关注点分离)
  • ✅ 更容易测试单个组件
  • ✅ 添加新功能的可扩展性更好
  • ✅ 服务层支持代码重用

无重大变更

两种实现方式的工作原理相同。旧文件保持向后兼容性,可以按照自己的节奏逐步淘汰。


�📝 许可证

麻省理工学院

🤝 贡献

  1. 分叉回购
  2. 创建特征分支
  3. 添加您的增强功能
  4. 提交拉取请求

为AI代理构建,以减少上下文窗口膨胀 🚀