import json import sys class McpProtocolHandler: protocol_version = '2025-06-18' server_name = 'fms-mcp-gateway' server_version = '0.1.0' def __init__(self, gateway_app): self.gateway_app = gateway_app self.initialized = False def handle_message(self, message): if not isinstance(message, dict): return self._error_response(None, -32600, 'Invalid Request') if 'id' in message: return self.handle_request(message) method = str(message.get('method') or '').strip() if method == 'notifications/initialized': self.initialized = True return None return None def handle_request(self, request): request_id = request.get('id') method = str(request.get('method') or '').strip() try: if method == 'initialize': self.initialized = True return self._success_response( request_id, { 'protocolVersion': self.protocol_version, 'capabilities': { 'tools': { 'listChanged': False, } }, 'serverInfo': { 'name': self.server_name, 'version': self.server_version, }, }, ) if method == 'tools/list': tools = [self._normalize_tool(tool) for tool in self.gateway_app.list_tools()] return self._success_response(request_id, {'tools': tools}) if method == 'tools/call': params = request.get('params') or {} tool_name = params.get('name') arguments = params.get('arguments') or {} tool_result = self.gateway_app.call_tool(tool_name, arguments) structured_content = tool_result.get('data') or {} return self._success_response( request_id, { 'content': [ { 'type': 'text', 'text': self._render_text(structured_content), } ], 'structuredContent': structured_content, 'isError': False, }, ) return self._error_response(request_id, -32601, 'Method not found: {0}'.format(method or '')) except Exception as exc: if method == 'tools/call': return self._success_response( request_id, { 'content': [ { 'type': 'text', 'text': str(exc), } ], 'isError': True, }, ) return self._error_response(request_id, -32000, str(exc)) def run_stdio(self, stdin=None, stdout=None): stdin = stdin or sys.stdin stdout = stdout or sys.stdout for raw_line in stdin: line = str(raw_line).strip() if not line: continue try: message = json.loads(line) except ValueError: response = self._error_response(None, -32700, 'Parse error') else: response = self.handle_message(message) if response is None: continue stdout.write(json.dumps(response, ensure_ascii=True) + '\n') if hasattr(stdout, 'flush'): stdout.flush() return 0 def _normalize_tool(self, tool): normalized = dict(tool) if 'input_schema' in normalized: normalized['inputSchema'] = normalized.pop('input_schema') return normalized @staticmethod def _render_text(structured_content): if not structured_content: return 'ok' columns = structured_content.get('columns') if isinstance(structured_content, dict) else None records = structured_content.get('records') if isinstance(structured_content, dict) else None if isinstance(columns, list) and isinstance(records, list): return McpProtocolHandler._render_table_like_text(structured_content, columns, records) if isinstance(records, list) and records: # Check if this is a track-like structure (no columns, but has records) first_record = records[0] if records else {} if 'status' in first_record and 'content' in first_record and 'time' in first_record: return McpProtocolHandler._render_track_like_text(structured_content, records) return json.dumps(structured_content, ensure_ascii=False) @staticmethod def _render_table_like_text(structured_content, columns, records): lines = [] summary = str(structured_content.get('summary') or '').strip() if summary: lines.append(summary) lines.append('表头共 {0} 列：'.format(len(columns))) for index, column in enumerate(columns, start=1): key = str(column.get('key') or '').strip() name = str(column.get('name') or key).strip() lines.append('{0}. {1} ({2})'.format(index, name, key)) if not records: tips = structured_content.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(str(tip) for tip in tips))) return '\n'.join(lines) for record_index, record in enumerate(records, start=1): lines.append('订单 {0}：'.format(record_index)) for column in columns: key = str(column.get('key') or '').strip() name = str(column.get('name') or key).strip() value = record.get(key, '') if isinstance(record, dict) else '' if value is None: value = '' lines.append('- {0}: {1}'.format(name, value)) tips = structured_content.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(str(tip) for tip in tips))) return '\n'.join(lines) @staticmethod def _render_track_like_text(structured_content, records): lines = [] summary = str(structured_content.get('summary') or '').strip() if summary: lines.append(summary) if not records: tips = structured_content.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(str(tip) for tip in tips))) return '\n'.join(lines) for record_index, record in enumerate(records, start=1): status = str(record.get('status') or '').strip() content = str(record.get('content') or '').strip() location = str(record.get('location') or '').strip() time = str(record.get('time') or '').strip() tracking_number = str(record.get('tracking_number') or '').strip() shipment_id = str(record.get('shipment_id') or '').strip() sub_track = int(record.get('sub_track') or 0) lines.append('轨迹 {0}：'.format(record_index)) lines.append('- 状态: {0}'.format(status or '无')) lines.append('- 内容: {0}'.format(content or '无')) if location: lines.append('- 地点: {0}'.format(location)) lines.append('- 时间: {0}'.format(time or '无')) if tracking_number: lines.append('- 快递单号: {0}'.format(tracking_number)) if shipment_id: lines.append('- Shipment ID: {0}'.format(shipment_id)) if sub_track: lines.append('- 子单轨迹: 是') tips = structured_content.get('tips') or [] if tips: lines.append('提示：{0}'.format('；'.join(str(tip) for tip in tips))) return '\n'.join(lines) @staticmethod def _success_response(request_id, result): return { 'jsonrpc': '2.0', 'id': request_id, 'result': result, } @staticmethod def _error_response(request_id, code, message): return { 'jsonrpc': '2.0', 'id': request_id, 'error': { 'code': code, 'message': message, }, }