使用说明
具有动态用户工具生成功能的MCP服务器
一个使用FastMCP的模型上下文协议(MCP)服务器实现,该服务器根据外部API中的用户权限动态生成工具。
特点/特性
- 动态工具生成工具是根据用户特定的模式,从外部API在运行时生成的
- 用户级权限每个经过身份验证的用户都会收到根据其权限定制的工具
- 智能缓存API模式和生成的工具均会以TTL进行缓存,以提高性能
- 优雅降级如果API不可用,则回退到静态工具
- 全面日志记录在INFO、DEBUG、WARNING和ERROR级别进行详细日志记录
- 线程安全多个用户并发请求得到安全处理
- 全局评分过滤Elasticsearch查询使用全局minScore阈值来控制结果质量
建筑学
组件
-
mcp_server.py- 主服务器入口点- 初始化FastMCP服务器
- 注册静态工具(加法、回显、乘法)
- 建立动态工具中间件
-
api_client.py- 外部API集成FormSchemaClient从外部API获取并解析架构FormSchemaCache缓存API响应并设置TTL- 将API字段定义转换为JSON Schema格式
-
dynamic_tool_manager.py- 工具生命周期管理DynamicToolManager管理每个用户的工具存储- 带有TTL的线程安全缓存
- 缓存统计和失效处理
-
tool_function_factory.py- 动态功能生成- 根据JSON模式创建带类型的异步函数
- 生成适当的
inspect.Signature用于FastMCP验证 - 将JSON模式类型映射到Python类型
-
tool_execution_handler.py- 工具执行后端ToolExecutionRouter将路由工具调用传递给相应的处理程序- 针对不同类型工具的专业操作人员
- 结构化的错误响应
-
dynamic_tool_middleware.py- FastMCP中间件- 拦截(或截距,根据上下文可具体确定)
list/tools请求: - 提取身份验证令牌
- 生成并缓存用户特定的工具
- 返回组合的静态+动态工具
- 拦截(或截距,根据上下文可具体确定)
安装
-
安装依赖项:
pip install -r requirements.txt -
配置API URL: 编辑
mcp_server.py并设置FORM_SCHEMA_API_URL到您的外部API端点:FORM_SCHEMA_API_URL = "http://172.16.11.131/api/module/request/form"
用法
启动服务器
python mcp_server.py
服务器将在 http://127.0.0.1:9092/mcp
认证
服务器期望通过HTTP进行身份验证 Authorization 头球
Authorization: Bearer <your-token>
该令牌用于:
- 从外部API获取用户特定的模式
- 生成用户有权访问的字段的工具
- 为后续请求缓存工具
可用工具
静态工具(始终可用)
add(a: int, b: int) -> int- 加法两个数echo(message: str) -> str- 回应一条信息multiply(a: int, b: int) -> int- 将两个数字相乘
动态工具(用户专用)
create_request(...)- 根据用户权限创建包含相应字段的请求- 参数根据外部API架构动态生成
- 每个用户根据其权限看到不同的参数
- 强制要求填写必填字段、枚举值和类型
它是如何运作的
请求流程
-
客户端发送
list/tools请求 (与……一起)带着Authorization: Bearer <token> -
中间件拦截请求:
- 从头部提取认证令牌
- 检查此用户是否已缓存工具
- 如果已缓存:返回缓存的工具
- 如果未缓存:则进行生成
-
工具生成:
- 使用用户的令牌从外部API获取模式
- 将API响应解析为JSON模式
- 创建具有适当签名的类型化异步函数
- 将函数转换为FastMCP
Tool对象 - 为未来的请求缓存该工具
-
回答:
- 返回静态工具和动态工具的列表
- 客户端仅能看到与其权限相关的工具
工具执行流程
-
客户来电
create_request工具 带有论据/理由 -
FastMCP验证参数 针对生成的函数签名
-
工具执行处理器:
- 接收经过验证的参数
- 处理请求
- 返回包含ID、时间戳和状态的结构化响应
配置
缓存TTL(Time To Live,生存时间)
默认情况下,模式缓存和工具缓存均使用5分钟的TTL(生存时间)。如需更改:
# In mcp_server.py
schema_client = FormSchemaClient(
api_url=FORM_SCHEMA_API_URL,
cache_ttl=300, # Change this value (seconds)
verbose=True
)
tool_manager = DynamicToolManager(cache_ttl_seconds=300) # Change this value
日志级别
更改日志详细程度:
# In mcp_server.py
logging.basicConfig(
level=logging.DEBUG, # Change to DEBUG, INFO, WARNING, or ERROR
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
测试
见 TEST_SCENARIOS.md 针对包括以下在内的全面测试场景:
- 用户特定工具的生成
- 缓存行为
- 工具执行
- 错误处理
- 验证执行
API架构格式
外部API应返回具有此结构的响应:
{
"fieldList": [
{
"name": "Subject",
"paramName": "subject",
"type": "TextFieldRest",
"required": true,
"description": "Request subject",
"groupIds": [1, 2],
"hidden": false,
"removed": false,
"inActive": false
}
]
}
支持的字段类型:
TextFieldRest→stringNumberFieldRest→numberDropDownFieldRest→string使用枚举(enum)MultiSelectDropDownFieldRest→array字符串的(集合/列表/等,具体根据上下文确定)- 并且更多(参见
api_client.py(用于完整映射)
安全考量
- 认证令牌在日志和响应中被截断
- 每次请求时都会验证令牌
- 无令牌 = 仅使用静态工具(优雅降级)
- FastMCP 在执行前验证所有工具参数
故障排除
服务器无法启动
- 检查所有依赖项是否已安装:
pip install -r requirements.txt - 验证Python版本是否为3.12或更高版本
- 检查日志以查找导入错误
工具未显示
- 验证
Authorization头部存在且格式正确 - 检查服务器日志以查找API获取错误
- 确保外部API可访问
- 验证用户在外部系统中是否有权限
缓存未更新
- 默认TTL(生存时间)为5分钟
- 手动清除缓存:呼叫
tool_manager.clear_all_tools()或者重启服务器 - 检查日志中的缓存命中/未命中消息
许可证
[此处填写您的许可证]
做出贡献
[在此处添加您的贡献指南]