当前位置：首页 > news >正文

bisheng 的 MCP服务器添加或系统集成

news 2025/10/22 8:48:52

我来帮您了解如何设计MCP服务器并添加到bisheng系统中。让我先查看现有的MCP相关代码。

[2 tools called]

现在让我查看MCP相关的API接口和配置：

[2 tools called]

基于我对bisheng系统中MCP架构的分析，我来为您详细介绍如何设计MCP服务器并添加到bisheng中。

MCP服务器设计与集成指南

1. MCP架构概述

bisheng系统已经集成了完整的MCP（Model Context Protocol）支持，包括：

客户端管理：ClientManager负责管理MCP连接
传输协议：支持SSE和STDIO两种传输方式
工具集成：自动将MCP工具转换为LangChain工具
API接口：提供完整的MCP工具管理API

2. MCP服务器配置格式

bisheng支持标准的MCP配置格式：

{"mcpServers": {"server_name": {"name": "服务器显示名称","description": "服务器描述","type": "stdio",  // 或 "sse""command": "python",  // STDIO模式"args": ["-m", "your_mcp_server"],"env": {"API_KEY": "${API_KEY}"}}}
}

支持的传输类型：

STDIO模式（推荐）

{"mcpServers": {"my_server": {"name": "我的MCP服务器","description": "自定义MCP服务器","command": "python","args": ["-m", "my_mcp_server"],"env": {"API_KEY": "your_api_key"}}}
}

SSE模式

{"mcpServers": {"my_server": {"name": "我的MCP服务器", "description": "自定义MCP服务器","url": "http://localhost:3000/sse"}}
}

3. 创建MCP服务器

3.1 使用Python创建MCP服务器

# my_mcp_server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent# 创建MCP服务器实例
server = Server("my-server")@server.list_tools()
async def list_tools() -> list[Tool]:"""列出可用工具"""return [Tool(name="get_weather",description="获取指定城市的天气信息",inputSchema={"type": "object","properties": {"city": {"type": "string","description": "城市名称"}},"required": ["city"]}),Tool(name="calculate",description="执行数学计算",inputSchema={"type": "object", "properties": {"expression": {"type": "string","description": "数学表达式"}},"required": ["expression"]})]@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:"""调用工具"""if name == "get_weather":city = arguments.get("city", "")# 这里调用实际的天气APIweather_info = f"{city}的天气：晴天，25°C"return [TextContent(type="text", text=weather_info)]elif name == "calculate":expression = arguments.get("expression", "")try:# 安全的数学计算（仅支持基本运算）result = eval(expression)return [TextContent(type="text", text=f"计算结果：{result}")]except Exception as e:return [TextContent(type="text", text=f"计算错误：{str(e)}")]else:return [TextContent(type="text", text=f"未知工具：{name}")]async def main():"""启动服务器"""async with stdio_server() as (read_stream, write_stream):await server.run(read_stream, write_stream, server.create_initialization_options())if __name__ == "__main__":asyncio.run(main())

3.2 使用Node.js创建MCP服务器

// my_mcp_server.js
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js';const server = new Server({name: "my-server",version: "1.0.0",},{capabilities: {tools: {},},}
);// 列出工具
server.setRequestHandler(ListToolsRequestSchema, async () => {return {tools: [{name: "get_weather",description: "获取指定城市的天气信息",inputSchema: {type: "object",properties: {city: {type: "string",description: "城市名称"}},required: ["city"]}}]};
});// 调用工具
server.setRequestHandler(CallToolRequestSchema, async (request) => {const { name, arguments: args } = request.params;if (name === "get_weather") {const city = args.city;return {content: [{type: "text",text: `${city}的天气：晴天，25°C`}]};}throw new Error(`未知工具：${name}`);
});// 启动服务器
async function main() {const transport = new StdioServerTransport();await server.connect(transport);
}main().catch(console.error);

4. 将MCP服务器添加到bisheng

4.1 通过API添加

# 1. 解析MCP配置
curl -X POST "http://localhost:4001/api/v1/assistant/mcp/tool_schema" \-H "Content-Type: application/json" \-d '{"file_content": "{\"mcpServers\":{\"my_server\":{\"name\":\"我的MCP服务器\",\"description\":\"自定义MCP服务器\",\"command\":\"python\",\"args\":[\"-m\",\"my_mcp_server\"]}}}"}'# 2. 测试MCP工具
curl -X POST "http://localhost:4001/api/v1/assistant/mcp/tool_test" \-H "Content-Type: application/json" \-d '{"openapi_schema": "{\"mcpServers\":{\"my_server\":{\"command\":\"python\",\"args\":[\"-m\",\"my_mcp_server\"]}}}","extra": "{\"name\":\"get_weather\"}","request_params": {"city": "北京"}}'

4.2 通过前端界面添加

进入工具管理页面
选择MCP工具类型
上传MCP配置文件
测试工具功能
保存配置

5. 高级MCP服务器功能

5.1 支持资源管理

@server.list_resources()
async def list_resources() -> list[Resource]:"""列出可用资源"""return [Resource(uri="file:///path/to/file.txt",name="示例文件",description="一个示例文本文件",mimeType="text/plain")]@server.read_resource()
async def read_resource(uri: str) -> str:"""读取资源内容"""if uri.startswith("file://"):file_path = uri[7:]  # 移除 file:// 前缀with open(file_path, 'r', encoding='utf-8') as f:return f.read()raise ValueError(f"不支持的URI: {uri}")

5.2 支持提示模板

@server.list_prompts()
async def list_prompts() -> list[Prompt]:"""列出可用提示模板"""return [Prompt(name="weather_prompt",description="天气查询提示模板",arguments=[PromptArgument(name="city",description="城市名称",required=True)])]@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[TextContent]:"""获取提示内容"""if name == "weather_prompt":city = arguments.get("city", "")return [TextContent(type="text",text=f"请查询{city}的天气信息，包括温度、湿度和天气状况。")]raise ValueError(f"未知提示：{name}")

6. 部署和配置

6.1 环境配置

# 安装MCP依赖
pip install mcp# 设置环境变量
export API_KEY="your_api_key"
export MCP_SERVER_PATH="/path/to/your/server"

6.2 系统集成

# 在bisheng中集成自定义MCP客户端
from bisheng.mcp_manage.manager import ClientManager
from bisheng.mcp_manage.langchain.tool import McpTool# 创建MCP客户端
client = await ClientManager.connect_mcp(client_type="stdio",command="python",args=["-m", "my_mcp_server"]
)# 获取工具列表
tools = await client.list_tools()# 转换为LangChain工具
for tool in tools:langchain_tool = McpTool.get_mcp_tool(name=tool.name,description=tool.description,mcp_client=client,mcp_tool_name=tool.name,arg_schema=tool.inputSchema)

7. 最佳实践

错误处理：在MCP服务器中实现完善的错误处理
安全性：验证输入参数，避免代码注入
性能优化：使用异步操作，避免阻塞
日志记录：添加详细的日志记录
测试覆盖：编写完整的单元测试和集成测试

8. 故障排除

常见问题：

连接失败：检查MCP服务器是否正常运行
工具未显示：验证MCP配置格式是否正确
调用失败：检查工具参数和返回值格式

调试方法：

# 启用调试模式
import logging
logging.basicConfig(level=logging.DEBUG)# 测试MCP连接
client = ClientManager.sync_connect_mcp_from_json(config)
tools = await client.list_tools()
print(f"可用工具：{[tool.name for tool in tools]}")