MCPeasy
_设置和自托管自己的多MCP服务器的最简单方法，具有可流式http传输、多客户端和密钥管理_
生产级多对映MCP服务器，使用基于API密钥的路由为不同的客户端提供不同的工具和配置。
建筑
- FastMCP 2.6：核心MCP实施如下https://gofastmcp.com/llms-full.txt
- 快速API：具有基于API密钥的URL路由的Web框架
- PostgreSQL：使用SQLAlchemy的多租户数据存储
- 流式HTTP：所有子服务器都提供流式传输
- 多租户技术：客户端可以具有多个API密钥，这些密钥具有特定于工具的配置
主要特点
- 多租户设计：客户端管理多个可旋转的API键
- 根据工具配置：每个客户端都可以不同地配置工具（例如，自定义电子邮件地址）
- 动态工具集：不同的客户得到不同的工具组合
- 工具自动发现：具有自动注册功能的模块化工具系统
- 自定义工具支持：在fork中添加特定于组织的工具，并使用命名空间目录
- 按资源配置：每个客户端都可以通过自定义设置访问不同的资源
- 动态资源集：不同的客户端获得不同的资源组合
- 资源自动发现：具有自动注册功能的模块化资源系统
- 增强的工具响应：多种内容类型（文本、JSON、markdown、文件）可实现最佳的LLM集成
- 基于环境的发现：用于工具/资源启用的简单环境变量配置
- 共享基础设施：跨服务器共享数据库、日志记录和配置
- 管理界面：基于Web的客户端和API密钥管理，带有CORE/CUSTOM工具源代码徽章
- 生产就绪：使用Neon数据库为Fly部署而构建
- 高性能：后台任务处理、请求超时、配置缓存和优化的数据库连接
快速开始
- 设置环境:
cp .env.example .env
# Edit .env with your database URL, admin password, and session secret- 使用Docker Compose启动所有服务 （推荐）：
docker-compose up- 访问服务:
http://localhost:8000 # Main API endpoint
http://localhost:3000 # Admin interface (on prod: https://yourdomain.com/admin). Log in with username 'superadmin', pass: your 'SUPERADMIN_PASSWORD' from .env
http://localhost:8080 # Database inspector (Adminer)就是这样！Docker Compose自动处理所有依赖关系、数据库设置和迁移。
API终点
GET /health-健康检查GET /admin-管理员登录页面GET /admin/clients-客户端管理仪表板POST /admin/clients-创建新客户端GET /admin/clients/{id}/keys-管理客户端的API密钥POST /admin/clients/{id}/keys-生成新的API密钥GET /admin/clients/{id}/tools-为客户端配置工具GET|POST /mcp/{api_key}-MCP端点（可流式传输）
客户和API密钥管理
创建客户
- 访问
/（开发中的localhost:3000）并使用超级管理员密码登录 - 创建具有名称和描述的客户端
- 为客户端生成API密钥
- 根据每个客户端的设置配置工具和资源
管理API密钥
- 每个客户端有多个密钥：生产、分期、开发关键
- 关键点旋转：在不丢失配置的情况下生成新密钥
- 到期管理：设置密钥的过期日期
- 安全删除：立即停用泄露的密钥
工具配置
每个客户端都必须明确配置访问它们的工具：
- 简单的工具:
echo,get_weather-单击“添加”启用（无需配置） - 可配置工具:
send_email-单击“配置”从地址、SMTP设置进行设置 - 每个客户端设置：相同的工具，每个客户端的配置不同
- 严格的访问控制：只有配置好的工具可见且可调用
可用工具
命名空间工具系统： 所有工具都组织在命名空间中，以更好地组织和避免冲突：
核心工具（命名空间： core/):
core/echo-用于测试的简单回声工具（无需配置）core/weather-天气信息（无需配置）core/send_email-发送电子邮件（必填：from_email，可选：smtp_server）core/datetime-日期和时间实用程序core/scrape-Web抓取功能core/youtube_lookup-YouTube视频信息
自定义工具（命名空间： {org}/):
myorg/send_invoice-自定义工具将位于此处- 可以在特定于组织的命名空间中添加自定义工具
- 每个部署都可以通过环境配置控制哪些工具可用
- 自定义工具在管理UI中显示紫色的“Custom”徽章，而蓝色的“CORE”徽章
工具发现：
- 使用
TOOLS=__all__自动发现并启用所有可用工具 - 或指定精确的工具：
TOOLS='core/echo,core/weather,myorg/send_invoice' - 目录结构：
src/tools/{namespace}/{tool_name}/tool.py
工具调用跟踪
所有工具执行都会在数据库中自动跟踪，以便进行监控和审计：
- 完整跟踪：输入参数、输出数据、执行时间和错误
- 每客户端日志记录：按客户端和API密钥跟踪使用模式
- 性能监控：执行时间跟踪（毫秒）
- 错误记录：工具调用失败，并显示详细的错误消息
- 自动：无需配置-所有工具调用都会透明记录
资源配置
每个客户端都必须明确配置资源才能访问它们：
- 简单资源：单击“添加”启用默认设置
- 可配置资源：单击“配置”以设置类别筛选器、文章限制、搜索权限
- 每个客户端设置：相同的资源，每个客户端的配置不同（例如，不同的类别访问）
- 严格的访问控制：只有配置的资源可见且可访问
可用资源
命名空间资源系统： 资源遵循与更好组织工具相同的命名空间模式：
自定义资源（命名空间： {org}/):
myorg/knowledge-命名空间资源示例- 可以在特定于组织的命名空间中添加自定义资源
- 每个部署都可以通过环境配置控制哪些资源可用
资源发现：
- 使用
RESOURCES=__all__自动发现并启用所有可用资源 - 或者指定确切的资源：
RESOURCES='myorg/product_catalog' - 目录结构：
src/resources/{namespace}/{resource_name}/resource.py
资源自动播种
当资源的表为空时，资源可以自动为初始数据播种，非常适合：
- 参考数据：国家、类别、产品目录
- 演示内容：样品文章、文件
- 初始配置：默认设置、预设
工作原理：
- 资源在首次初始化时检查其表是否为空
- 如果为空，则从配置的源（CSV/JSON文件或URL）加载种子数据
- 使用正确的字段映射将数据插入数据库
- 只运行一次-后续初创公司跳过种子
设置示例：
class ProductCatalogResource(BaseResource):
name = "myorg/products"
seed_source = "seeds/initial_products.csv" # Local file
# seed_source = "https://cdn.myorg.com/products.csv" # Or remote URL
async def _get_model_class(self):
from .models import Product
return Product支持的格式：
- CSV文件：列名与模型字段匹配，空字符串变为NULL
- 对象符号：以字段名为关键字的对象数组
- 远程URL：从CDN或API获取种子数据
种子文件示例：
# seeds/products.csv
id,name,category,price,description
1,Widget A,widgets,29.99,Premium widget
2,Widget B,widgets,39.99,Deluxe widget// seeds/products.json
[
{"id": 1, "name": "Widget A", "category": "widgets", "price": 29.99},
{"id": 2, "name": "Widget B", "category": "widgets", "price": 39.99}
]配置
环境变量
DATABASE_URL=postgresql://user:pass@host:port/db
PORT=8000
SESSION_SECRET=your_secure_session_key_here
SUPERADMIN_PASSWORD=your_secure_password
# Tool and resource discovery
TOOLS=__all__ # Enable all discovered tools, or list specific ones like 'core/echo,m38/calculator'
RESOURCES=__all__ # Enable all discovered resources, or list specific ones like 'knowledge'
# Tool execution queue configuration
TOOL_MAX_WORKERS=20 # Max concurrent tool executions (default: 20)
TOOL_QUEUE_SIZE=200 # Max queued requests (default: 200)更多信息请参见.env.example
多租户架构
该系统使用三个主要实体：
- 客户：具有UUID标识符的组织或用户（例如“ACME Corp”）
- API密钥：每个客户有多个可旋转的钥匙
- 工具配置：每个客户端的工具设置以JSON格式存储，具有严格的访问控制
- 资源配置：每个客户端的资源设置以JSON格式存储，具有严格的访问控制
自定义工具开发
MCPeasy支持使用简化的命名空间目录结构添加特定于组织的工具。分叉此存储库时，您可以直接添加自定义工具，而无需担心合并冲突。
快速自定义工具设置
- 分叉存储库：创建自己的mcpeasy叉子
- 创建命名空间目录:
mkdir -p src/tools/yourorg - 添加您的工具：创建
src/tools/yourorg/yourtool/tool.py使用您的工具实现 - 自动发现：工具自动发现为
yourorg/yourtool - 配置环境:
- 使用 TOOLS=__all__ 自动启用所有工具 - 或指定： TOOLS='core/echo,yourorg/yourtool'
- 为客户端启用：使用管理UI为每个客户端配置工具
- 保持更新：需要时从mcpeasy主分支提取上游更改
目录结构
src/tools/
├── core/ # Core mcpeasy tools
│ ├── echo/
│ ├── weather/
│ └── send_email/
├── m38/ # Example custom namespace
│ └── calculator/
└── yourorg/ # Your organization's tools
├── invoice_generator/
├── crm_integration/
└── custom_reports/增强的工具响应类型
自定义工具支持多种内容类型，以实现最佳的LLM集成：
# Structured data (recommended for LLM processing)
return ToolResult.json({"result": 42, "status": "success"})
# Human-readable text
return ToolResult.text("Operation completed successfully!")
# Markdown formatting
return ToolResult.markdown("# Success\n\n**Result**: 42")
# File references
return ToolResult.file("s3://bucket/report.pdf", mime_type="application/pdf")
# Error handling
return ToolResult.error("Invalid operation: division by zero")在工具中运行同步代码
如果您的自定义工具需要运行同步（阻塞）代码，请使用 asyncio.to_thread() 为了避免阻塞异步事件循环：
import asyncio
from src.tools.base import BaseTool, ToolResult
class MyCustomTool(BaseTool):
@property
def name(self) -> str:
return "my_custom_tool"
async def execute(self, arguments: dict, config: dict = None) -> ToolResult:
# For CPU-bound or blocking I/O operations
result = await asyncio.to_thread(self._blocking_operation, arguments)
return ToolResult.json(result)
def _blocking_operation(self, arguments: dict):
# This runs in a thread pool, safe to block
import time
time.sleep(5) # Example: blocking operation
return {"processed": True, "data": arguments}重要：切勿直接在 execute() 方法，因为它会阻塞整个事件循环并影响其他工具的执行。
定制资源开发
MCPeasy支持添加具有自动数据种子功能的特定于组织的资源。就像使用工具一样，将您的自定义资源直接添加到您的fork中。
快速自定义资源设置
- 在你的叉子里：导航到您的mcpeasy叉子
- 创建命名空间目录:
mkdir -p src/resources/yourorg - 添加您的资源：创建
src/resources/yourorg/yourresource/resource.py随着实施 - 自动发现：资源自动发现为
yourorg/yourresource - 配置环境:
- 使用 RESOURCES=__all__ 自动启用所有资源 - 或指定： RESOURCES='knowledge,yourorg/yourresource'
- 可选播种：添加
seed_source和seeds/初始数据目录 - 为客户端启用：使用管理UI为每个客户端配置资源
目录结构
src/resources/
├── knowledge/ # Core resources (simple)
├── myorg/ # Namespaced resources
│ └── knowledge/
│ ├── resource.py
│ ├── models.py
│ └── seeds/
│ ├── articles.csv
│ └── categories.json
└── yourorg/ # Your organization's resources
├── product_catalog/
│ ├── resource.py
│ ├── models.py
│ └── seeds/
│ └── products.csv
└── customer_data/
├── resource.py
└── models.py具有自动种子的自定义资源
from src.resources.base import BaseResource
from src.resources.types import MCPResource, ResourceContent
class ProductCatalogResource(BaseResource):
name = "yourorg/products"
description = "Product catalog with pricing and inventory"
uri_scheme = "products"
# Optional: Auto-seed when table is empty
seed_source = "seeds/initial_products.csv"
async def _get_model_class(self):
from .models import Product
return Product
async def list_resources(self, config=None):
# Implementation with client-specific filtering
pass
async def read_resource(self, uri: str, config=None):
# Implementation with access control
pass模板和文档
- 模板：中的完整工具/资源模板
templates/带有自动种子设定示例的目录 - 最佳实践：示例显示了正确的依赖关系管理、配置和数据种子
- 命名空间组织：核心工具和自定义工具/资源之间的清晰分离
- 环境变量发现：简单的工具和资源配置
- 种子数据示例：包括CSV和JSON种子文件模板
发展
Docker开发（推荐）
# Start all services with live reload
docker-compose up
# Access services:
# - App: http://localhost:8000
# - Admin: http://localhost:3000
# - Database Inspector: http://localhost:8080前端和后端的实时重新加载
数据库检查器（管理员）
当使用Docker Compose运行时，Adminer提供了一个轻量级的web界面来检查PostgreSQL数据库：
- 统一资源定位符:
http://localhost:8080 - 登录信息:
- 服务器： db - 用户名： postgres - 密码： postgres - 数据库： mcp
特性:
- 浏览所有表（客户端、api_keys、工具配置、资源配置、工具调用）
- 查看表数据和关系
- 运行SQL查询
- 导出数据
- 监视数据库架构更改
- 分析工具使用模式和性能指标
地方发展
- 依赖项：管理方式
uv - 编码结构：采用SQLAlchemy模型、会话认证、管理UI的模块化设计
- 数据库：带有异步SQLAlchemy和Alembic迁移的PostgreSQL
- 认证：使用安全Cookie的基于会话的管理员身份验证
- 迁移：使用Alembic自动进行数据库迁移
- 测试：使用自动重新加载运行开发服务器
测试MCP端点
使用MCP检查器（推荐）
- 获取令牌URL：从管理仪表板中，复制您的令牌的MCP URL
- 安装检查员:
npx @modelcontextprotocol/inspector - 开放式检查员：参观http://localhost:6274浏览器中（如果需要，包括代理身份验证，请按照检查器启动时的说明进行操作）
- 添加服务器：输入您的MCP URL：
http://localhost:8000/mcp/{token} - 配置工具和资源：在管理界面中，为您的客户端添加/配置工具和资源
- 测试功能：单击已配置的工具和资源进行测试（未配置的项目不会显示）
✅ 已验证工作：MCP检查器成功连接并仅显示配置的工具和资源！
手动测试
# Test capability discovery
curl http://localhost:8000/mcp/{your_api_key}
# Test echo tool (no configuration needed)
curl -X POST http://localhost:8000/mcp/{your_api_key} \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "echo",
"arguments": {"message": "Hello MCP!"}
}
}'
# Test send_email tool (uses client-specific configuration)
curl -X POST http://localhost:8000/mcp/{your_api_key} \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "send_email",
"arguments": {
"to": "user@example.com",
"subject": "Test",
"body": "This uses my configured from address!"
}
}
}'
# Test knowledge resource (uses client-specific configuration)
curl -X POST http://localhost:8000/mcp/{your_api_key} \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "resources/read",
"params": {
"uri": "knowledge://search?q=api"
}
}'数据库迁移
系统使用Alembic进行数据库迁移 Docker启动时的自动执行 以获得最佳的开发人员体验。
迁移工作流（简化）
# 1. Create a new migration after making model changes
./migrate.sh create "add user preferences table"
# 2. Restart the app (migrations apply automatically)
docker-compose restart app
# That's it! No manual migration commands needed.可用的迁移命令
这 ./migrate.sh 脚本提供所有迁移功能：
# Create new migration (auto-starts database if needed)
./migrate.sh create "migration message"
# Apply pending migrations manually (optional)
./migrate.sh upgrade
# Check current migration status
./migrate.sh status
# View migration history
./migrate.sh history运作原理
- 发展：使用
./migrate.sh create "message"生成迁移文件 - 自动应用：Docker容器启动时，迁移会自动运行
- 无手动步骤：Docker容器句柄
alembic upgrade head启动时 - 数据库相关性：Docker在运行迁移之前等待数据库健康检查
- 数据载体安装：迁移文件可以通过卷装载立即在容器中使用
模型组织
模型按域组织在单独的文件中：
src/models/base.py-SQLAlchemy基类src/models/client.py-客户端和APIKey模型src/models/configuration.py-工具和资源配置src/models/knowledge.py-知识库模型src/models/tool_call.py-工具调用跟踪和审计
迁移工作流
- 更改模型 在相应的模型文件中
- 生成迁移：系统会自动检测更改并创建迁移文件
- 审查迁移：检查中生成的SQL
src/migrations/versions/ - 部署：迁移在生产环境中启动时自动运行
生产迁移行为
- ✅ 自动执行：迁移在应用程序启动时运行
- ✅ 安全推出：迁移失败会阻止应用程序启动
- ✅ 版本跟踪：数据库跟踪当前迁移状态
- ✅ 幂等：多次运行安全
性能和可扩展性
该系统针对生产工作负载进行了优化，并进行了多项性能增强：
- 基于队列的执行：可配置工作池的有限并发性可防止服务器过载
- 公平调度：FIFO队列确保在流量突发期间为所有客户端提供服务
- 后台处理：工具调用日志记录移至后台任务，以加快响应时间
- 延长超时时间：3分钟超时支持长时间运行的工具（可配置）
- 配置缓存：5分钟TTL缓存减少了配置查找的数据库查询
- 连接池：通过预ping验证优化PostgreSQL连接管理
- 多工人设置：2名工作人员针对Fly.io部署进行了优化，具有自动回收功能
- 队列监视：实时队列指标可在
/metrics/queue端点
队列配置
控制工具执行并发性和队列行为：
# Environment variables for .env
TOOL_MAX_WORKERS=20 # Concurrent tool executions (default: 20)
TOOL_QUEUE_SIZE=200 # Maximum queued requests (default: 200)
# Recommended settings by server size:
# Small servers: TOOL_MAX_WORKERS=5, TOOL_QUEUE_SIZE=50
# Production: TOOL_MAX_WORKERS=50, TOOL_QUEUE_SIZE=500队列监视
# Check queue health and capacity
curl http://localhost:8000/metrics/queue
# Response includes:
{
"queue_depth": 3, # Current requests waiting
"max_workers": 20, # Maximum concurrent executions
"max_queue_size": 200, # Maximum queue capacity
"workers_started": 20, # Number of active workers
"is_started": true # Queue system status
}部署
- 平台：建议使用Fly.io.NB进行部署！在某些情况下（例如，如果连接到此的MCP客户端在cloudflare workers内部运行，则应设置
force_https = false在fly.toml中，否则您可能会在MCP客户端遇到无休止的重定向问题） - 数据库：任何postgres都可以，在Neon PostgreSQL上通过自动迁移进行测试
- 环境：生产准备就绪，具有适当的错误处理和迁移安全性
- 工人：2名Uvicorn员工，要求回收1000个内存，以实现最佳内存管理