mcp/mcp_protocol.py

import json
import logging
import sys
import uuid

from services.output_presenter import OutputPresenter
from services.diagnostic_event import RequestDiagnosticEmitter
from services.diagnostic_reporter import NullDiagnosticReporter


logger = logging.getLogger(__name__)


class McpProtocolHandler:
    protocol_version = '2025-06-18'
    server_name = 'fms-mcp-gateway'
    server_version = '0.1.0'
    output_presenter = OutputPresenter()

    @staticmethod
    def server_capabilities():
        return {
            'tools': {
                'listChanged': False,
            },
            'resources': {
                'listChanged': False,
            },
        }

    def __init__(self, gateway_app, reporter=None):
        self.gateway_app = gateway_app
        self.reporter = reporter or NullDiagnosticReporter()
        self.initialized = False

    def handle_message(self, message):
        if not isinstance(message, dict):
            return self._error_response(
                None,
                -32600,
                'Invalid Request',
                self._build_trace_request_id(),
            )
        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()
        tool_name = ''
        trace_request_id = self._build_trace_request_id()
        emitter = RequestDiagnosticEmitter(self.reporter, trace_request_id)
        emitter.emit(
            stage='request_ingress',
            status='started',
            event_code='REQUEST_RECEIVED',
            context={
                'jsonrpc_method': method or 'unknown',
                'transport': 'stdio',
            },
        )
        backend_started = False
        try:
            if method == 'initialize':
                emitter.emit(
                    stage='protocol_validation',
                    status='succeeded',
                    event_code='PROTOCOL_VALIDATION_COMPLETED',
                    context={
                        'jsonrpc_method': method,
                        'transport': 'stdio',
                    },
                )
                self.initialized = True
                return self._success_response(
                    request_id,
                    {
                        'protocolVersion': self.protocol_version,
                        'capabilities': self.server_capabilities(),
                        'serverInfo': {
                            'name': self.server_name,
                            'version': self.server_version,
                        },
                    },
                )
            if method == 'tools/list':
                emitter.emit(
                    stage='protocol_validation',
                    status='succeeded',
                    event_code='PROTOCOL_VALIDATION_COMPLETED',
                    context={
                        'jsonrpc_method': method,
                        'transport': 'stdio',
                    },
                )
                tools = [
                    self._normalize_tool(tool)
                    for tool in self.gateway_app.list_tools(
                        request_id=trace_request_id,
                    )
                ]
                return self._success_response(request_id, {'tools': tools})
            if method == 'resources/list':
                emitter.emit(
                    stage='protocol_validation',
                    status='succeeded',
                    event_code='PROTOCOL_VALIDATION_COMPLETED',
                    context={
                        'jsonrpc_method': method,
                        'transport': 'stdio',
                    },
                )
                resources = []
                if hasattr(self.gateway_app, 'list_resources'):
                    resources = list(self.gateway_app.list_resources() or [])
                return self._success_response(
                    request_id,
                    {'resources': resources},
                )
            if method == 'resources/read':
                params = request.get('params') or {}
                uri = ''
                if isinstance(params, dict):
                    uri = str(params.get('uri') or '').strip()
                emitter.emit(
                    stage='protocol_validation',
                    status='succeeded',
                    event_code='PROTOCOL_VALIDATION_COMPLETED',
                    context={
                        'jsonrpc_method': method,
                        'transport': 'stdio',
                    },
                )
                resource = None
                if uri and hasattr(self.gateway_app, 'read_resource'):
                    resource = self.gateway_app.read_resource(uri)
                if not isinstance(resource, dict):
                    return self._error_response(
                        request_id,
                        -32602,
                        'Invalid params',
                        trace_request_id,
                    )
                return self._success_response(
                    request_id,
                    {'contents': [resource]},
                )
            if method == 'tools/call':
                params = request.get('params') or {}
                if not isinstance(params, dict):
                    raise ValueError('tool parameters must be an object')
                raw_tool_name = params.get('name')
                if not isinstance(raw_tool_name, str) or not raw_tool_name.strip():
                    emitter.emit(
                        stage='protocol_validation',
                        status='failed',
                        event_code='PARAM_VALIDATION_FAILED',
                        context={
                            'jsonrpc_method': method,
                            'jsonrpc_code': -32602,
                            'transport': 'stdio',
                        },
                    )
                    return self._error_response(
                        request_id,
                        -32602,
                        'Invalid params',
                        trace_request_id,
                    )
                tool_name = raw_tool_name.strip()
                arguments = params.get('arguments') or {}
                emitter.emit(
                    stage='protocol_validation',
                    status='succeeded',
                    event_code='PROTOCOL_VALIDATION_COMPLETED',
                    tool_code=tool_name,
                    context={
                        'jsonrpc_method': method,
                        'transport': 'stdio',
                    },
                )
                backend_started = True
                emitter.emit(
                    stage='backend_call',
                    status='started',
                    event_code='BACKEND_CALL_STARTED',
                    tool_code=tool_name,
                    context={'transport': 'stdio'},
                )
                tool_result = self.gateway_app.call_tool(
                    tool_name,
                    arguments,
                    request_id=trace_request_id,
                )
                backend_started = False
                emitter.emit(
                    stage='backend_call',
                    status='succeeded',
                    event_code='BACKEND_CALL_COMPLETED',
                    tool_code=tool_name,
                    response_code=(
                        tool_result.get('code')
                        if isinstance(tool_result, dict)
                        else None
                    ),
                    context={'transport': 'stdio'},
                )
                try:
                    response = self._tool_call_response(
                        request_id,
                        tool_name,
                        tool_result,
                    )
                except Exception:
                    emitter.emit(
                        stage='response_safety',
                        status='failed',
                        event_code='RESPONSE_SAFETY_REJECTED',
                        tool_code=tool_name,
                        context={'transport': 'stdio'},
                    )
                    raise
                emitter.emit(
                    stage='response_safety',
                    status='succeeded',
                    event_code='RESPONSE_SAFETY_COMPLETED',
                    tool_code=tool_name,
                    context={'transport': 'stdio'},
                )
                return response
            return self._error_response(
                request_id,
                -32601,
                'Method not found: {0}'.format(method or '<empty>'),
                trace_request_id,
            )
        except Exception as exc:
            if method == 'tools/call':
                diagnostic_reason = (
                    'PARAM_VALIDATION_FAILED'
                    if isinstance(exc, ValueError)
                    else 'UNEXPECTED_EXCEPTION'
                )
                logger.error(
                    'MCP stdio tool request failed',
                    extra={
                        'request_id': trace_request_id,
                        'jsonrpc_id': request_id,
                        'protocol_method': method,
                        'tool_code': tool_name,
                        'response_code': 'MCP_9001',
                        'diagnostic_reason': diagnostic_reason,
                        'exception_class': exc.__class__.__name__,
                    },
                )
                if backend_started:
                    emitter.emit(
                        stage='backend_call',
                        status='failed',
                        event_code=diagnostic_reason,
                        tool_code=tool_name or None,
                        response_code='MCP_9001',
                        context={'transport': 'stdio'},
                    )
                return self._tool_exception_response(
                    request_id,
                    tool_name,
                    exc,
                    trace_request_id,
                )
            logger.error(
                "MCP stdio request failed",
                extra={
                    'request_id': trace_request_id,
                    'jsonrpc_id': request_id,
                    'protocol_method': method,
                    'tool_code': tool_name,
                    'protocol_code': -32000,
                    'diagnostic_reason': 'UNEXPECTED_EXCEPTION',
                    'exception_class': exc.__class__.__name__,
                },
            )
            return self._error_response(
                request_id,
                -32000,
                'Gateway request failed. Please try again later.',
                trace_request_id,
            )

    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',
                    self._build_trace_request_id(),
                )
            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_name, tool_result):
        if tool_name == 'query_order':
            return cls._legacy_tool_call_response(request_id, tool_result)

        presented = cls.output_presenter.present(tool_name, tool_result)
        return cls._presented_response(request_id, presented)

    @classmethod
    def _tool_exception_response(
        cls,
        request_id,
        tool_name,
        exception,
        trace_request_id='',
    ):
        if tool_name == 'query_order':
            return cls._success_response(
                request_id,
                {
                    'content': [{
                        'type': 'text',
                        'text': str(exception),
                    }],
                    'isError': True,
                },
            )

        presented = cls.output_presenter.present_exception(
            tool_name,
            exception,
        )
        if trace_request_id:
            presented['meta'] = {'request_id': trace_request_id}
        return cls._presented_response(request_id, presented)

    @classmethod
    def _presented_response(cls, request_id, presented):
        result = {
            'content': [{
                'type': 'text',
                'text': presented['text'],
            }],
            'structuredContent': presented['structured_content'],
            'isError': presented['is_error'],
        }
        if presented['meta']:
            result['_meta'] = presented['meta']
        return cls._success_response(request_id, result)

    @classmethod
    def _legacy_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 _build_trace_request_id():
        return 'rq_stdio_{0}'.format(uuid.uuid4().hex[:16])

    @staticmethod
    def _error_response(request_id, code, message, trace_request_id=''):
        error = {
            'code': code,
            'message': message,
        }
        if trace_request_id:
            error['data'] = {'request_id': trace_request_id}
        return {
            'jsonrpc': '2.0',
            'id': request_id,
            'error': error,
        }