"""
MCP 服务器核心模块
实现 Model Context Protocol 的服务器端逻辑
"""
import json
import logging
from typing import Dict, Any, List, Optional
from datetime import datetime
from config import config
from tools import MCPTools, get_tool_schema
# 配置日志
logging.basicConfig(level=getattr(logging, config.LOG_LEVEL))
logger = logging.getLogger(__name__)
class MCPServer:
"""MCP 服务器实现类"""
def __init__(self):
self.tools = MCPTools()
self.resources = {}
self.tool_schemas = get_tool_schema()
self._initialize_resources()
def _initialize_resources(self):
"""初始化资源"""
self.resources = {
"config": {
"uri": "config://server",
"name": "Server Configuration",
"description": "服务器配置信息",
"mimeType": "application/json"
},
"status": {
"uri": "status://server",
"name": "Server Status",
"description": "服务器状态信息",
"mimeType": "application/json"
},
"logs": {
"uri": "logs://server",
"name": "Server Logs",
"description": "服务器日志信息",
"mimeType": "text/plain"
}
}
def get_server_info(self) -> Dict[str, Any]:
"""获取服务器信息"""
return {
"name": config.MCP_SERVER_NAME,
"version": config.MCP_SERVER_VERSION,
"protocol_version": "2024-11-05",
"capabilities": {
"tools": {},
"resources": {},
"logging": {}
}
}
def list_tools(self) -> Dict[str, Any]:
"""列出所有可用工具"""
try:
return {
"tools": self.tool_schemas
}
except Exception as e:
logger.error(f"Error listing tools: {e}")
return {
"error": f"Failed to list tools: {str(e)}"
}
def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""
调用指定工具
Args:
name: 工具名称
arguments: 工具参数
Returns:
工具执行结果
"""
try:
logger.info(f"Calling tool: {name} with arguments: {arguments}")
# 检查工具是否存在
if not hasattr(self.tools, name):
return {
"error": f"Tool '{name}' not found",
"available_tools": [schema["name"] for schema in self.tool_schemas]
}
# 获取工具方法并调用
tool_method = getattr(self.tools, name)
result = tool_method(**arguments)
logger.info(f"Tool {name} executed successfully")
return {
"content": [
{
"type": "text",
"text": json.dumps(result, indent=2, ensure_ascii=False)
}
]
}
except TypeError as e:
logger.error(f"Invalid arguments for tool {name}: {e}")
return {
"error": f"Invalid arguments for tool '{name}': {str(e)}"
}
except Exception as e:
logger.error(f"Error calling tool {name}: {e}")
return {
"error": f"Tool execution failed: {str(e)}"
}
def list_resources(self) -> Dict[str, Any]:
"""列出所有可用资源"""
try:
resources_list = []
for resource_id, resource_info in self.resources.items():
resources_list.append({
"uri": resource_info["uri"],
"name": resource_info["name"],
"description": resource_info.get("description", ""),
"mimeType": resource_info.get("mimeType", "text/plain")
})
return {
"resources": resources_list
}
except Exception as e:
logger.error(f"Error listing resources: {e}")
return {
"error": f"Failed to list resources: {str(e)}"
}
def read_resource(self, uri: str) -> Dict[str, Any]:
"""
读取指定资源
Args:
uri: 资源URI
Returns:
资源内容
"""
try:
logger.info(f"Reading resource: {uri}")
# 根据URI类型返回相应内容
if uri == "config://server":
content = json.dumps(config.get_mcp_server_info(), indent=2, ensure_ascii=False)
mime_type = "application/json"
elif uri == "status://server":
status_info = {
"status": "running",
"timestamp": datetime.now().isoformat(),
"uptime": "N/A", # 在实际应用中可以计算真实的运行时间
"tools_available": len(self.tool_schemas),
"resources_available": len(self.resources)
}
content = json.dumps(status_info, indent=2, ensure_ascii=False)
mime_type = "application/json"
elif uri == "logs://server":
# 在实际应用中，这里应该返回真实的日志内容
log_content = f"""[{datetime.now().isoformat()}] INFO: MCP Server started
[{datetime.now().isoformat()}] INFO: Tools loaded: {len(self.tool_schemas)}
[{datetime.now().isoformat()}] INFO: Resources initialized: {len(self.resources)}
"""
content = log_content
mime_type = "text/plain"
else:
return {
"error": f"Resource not found: {uri}",
"available_resources": list(self.resources.keys())
}
return {
"contents": [
{
"uri": uri,
"mimeType": mime_type,
"text": content
}
]
}
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
return {
"error": f"Failed to read resource: {str(e)}"
}
def ping(self) -> Dict[str, Any]:
"""健康检查"""
return {
"status": "ok",
"timestamp": datetime.now().isoformat(),
"server": config.MCP_SERVER_NAME,
"version": config.MCP_SERVER_VERSION
}
# 全局MCP服务器实例
mcp_server = MCPServer()
