mcp/mcp_protocol.py

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)
                return self._tool_call_response(request_id, tool_result)
            return self._error_response(request_id, -32601, 'Method not found: {0}'.format(method or '<empty>'))
        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

    @classmethod
    def _tool_call_response(cls, request_id, tool_result):
        if not isinstance(tool_result, dict):
            raise RuntimeError('invalid tool response')

        raw_code = tool_result.get('code')
        code = str(raw_code).strip() if raw_code is not None else ''
        message = str(tool_result.get('msg') or '').strip()
        data = tool_result.get('data')
        meta = tool_result.get('meta')

        if code in ('MCP_0000', '0'):
            if isinstance(data, dict):
                structured_content = dict(data)
            elif data:
                structured_content = {'data': data}
            else:
                structured_content = {}
            if isinstance(meta, dict) and meta:
                structured_content['meta'] = dict(meta)

            return cls._success_response(request_id, {
                'content': [{
                    'type': 'text',
                    'text': cls._render_text(structured_content),
                }],
                'structuredContent': structured_content,
                'isError': False,
            })

        error_content = {
            'code': code or 'MCP_9001',
            'msg': message or 'tool call failed',
        }
        if data not in (None, [], {}):
            error_content['data'] = data
        if isinstance(meta, dict) and meta:
            error_content['meta'] = dict(meta)

        return cls._success_response(request_id, {
            'content': [{
                'type': 'text',
                'text': '{0}: {1}'.format(
                    error_content['code'],
                    error_content['msg'],
                ),
            }],
            'structuredContent': error_content,
            'isError': True,
        })

    @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,
            },
        }