From ea90128b205b6e2a4d89170f007b5bacccf75012 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=A5=BF=E7=93=9C=C2=B0?= Date: Wed, 17 Jun 2026 15:09:44 +0800 Subject: [PATCH] 1 --- backend/mcpServer.py | 61 ++++++++++++++++++++++++++++++++++++++------ 1 file changed, 53 insertions(+), 8 deletions(-) diff --git a/backend/mcpServer.py b/backend/mcpServer.py index d1d8d18c..e58dd1e6 100644 --- a/backend/mcpServer.py +++ b/backend/mcpServer.py @@ -1,12 +1,49 @@ - +import argparse import os import time +import mcp.types as mt from fastmcp import FastMCP +from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext from main import application from fastmcp.server.providers.openapi import MCPType +from fastmcp.tools.base import ToolResult from config.main import config + +def parse_args(): + parser = argparse.ArgumentParser(description='Dcalc MCP server') + parser.add_argument( + '--tool-result-mode', + choices=['structured', 'both'], + default='structured', + help='控制工具返回内容:structured 仅保留 structuredContent;both 保留默认双份。', + ) + args, _ = parser.parse_known_args() + return args + + +class ToolResultModeMiddleware(Middleware): + def __init__(self, mode: str): + self.mode = mode + + async def on_call_tool( + self, + context: MiddlewareContext[mt.CallToolRequestParams], + call_next: CallNext[mt.CallToolRequestParams, ToolResult], + ) -> ToolResult: + result = await call_next(context) + if result.structured_content is None: + return result + if self.mode == 'structured': + return ToolResult( + content=[], + structured_content=result.structured_content, + meta=result.meta, + ) + return result + + def custom_route_mapper(route, mcp_type: MCPType) -> MCPType | None: """Advanced route type mapping.""" # Convert all admin routes to tools regardless of HTTP method @@ -15,14 +52,22 @@ def custom_route_mapper(route, mcp_type: MCPType) -> MCPType | None: else: return MCPType.EXCLUDE -mcp = FastMCP.from_fastapi( - app=application, - name="Dcalc_MCP", - instructions="DNF相关的MCP服务", - route_map_fn=custom_route_mapper, -) + +def create_mcp(tool_result_mode: str = 'structured') -> FastMCP: + mcp = FastMCP.from_fastapi( + app=application, + name="Dcalc_MCP", + instructions="DNF相关的MCP服务", + route_map_fn=custom_route_mapper, + ) + mcp.add_middleware(ToolResultModeMiddleware(tool_result_mode)) + return mcp os.environ['TZ'] = 'Asia/Shanghai' time.tzset() -mcp.run(transport="streamable-http", host="0.0.0.0", port=config.MCPPORT) + +if __name__ == '__main__': + args = parse_args() + mcp = create_mcp(args.tool_result_mode) + mcp.run(transport="streamable-http", host="0.0.0.0", port=config.MCPPORT)