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)
                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 '<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=False) + '\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'
        return json.dumps(structured_content, ensure_ascii=False)

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