mcp/public_gateway.py

import logging
import time
import uuid

from constants import DEVICE_INVALID_MESSAGE
from tools.list_order_filter_options import ListOrderFilterOptionsTool
from tools.list_outbound_filter_options import ListOutboundFilterOptionsTool
from tools.export_pending_outbound_orders import ExportPendingOutboundOrdersTool
from tools.export_out_of_province_port_data import (
    ExportOutOfProvincePortDataTool,
)
from tools.list_pending_outbound_export_filter_options import (
    ListPendingOutboundExportFilterOptionsTool,
)
from tools.query_order import QueryOrderTool
from tools.query_customs_declaration_files import (
    QueryCustomsDeclarationFilesTool,
)
from tools.query_order_exact import QueryOrderExactTool
from tools.query_order_detail import QueryOrderDetailTool
from tools.query_outbound_detail import QueryOutboundDetailTool
from tools.query_outbound_list import QueryOutboundListTool
from tools.query_track import QueryTrackTool
from utils.security import hash_gateway_session_id


logger = logging.getLogger(__name__)


class PublicGatewayApp:
    def __init__(self, session_store, api_client, auth_client=None):
        self.session_store = session_store
        self.api_client = api_client
        self._tools = {
            'query_order': QueryOrderTool(api_client=None),
            'query_track': QueryTrackTool(api_client=None),
            'query_order_exact': QueryOrderExactTool(api_client=None),
            'query_order_detail': QueryOrderDetailTool(api_client=None),
            'query_customs_declaration_files':
                QueryCustomsDeclarationFilesTool(api_client=None),
            'query_outbound_list': QueryOutboundListTool(api_client=None),
            'query_outbound_detail': QueryOutboundDetailTool(api_client=None),
            'list_outbound_filter_options': ListOutboundFilterOptionsTool(
                api_client=None
            ),
            'list_order_filter_options': ListOrderFilterOptionsTool(
                api_client=None
            ),
            'export_pending_outbound_orders': ExportPendingOutboundOrdersTool(
                api_client=None
            ),
            'export_out_of_province_port_data':
                ExportOutOfProvincePortDataTool(api_client=None),
            'list_pending_outbound_export_filter_options':
                ListPendingOutboundExportFilterOptionsTool(api_client=None),
        }

    def registered_tool_names(self):
        return tuple(self._tools.keys())

    def _require_session(self, gateway_session_id, diagnostic_emitter=None):
        session = self.session_store.get(gateway_session_id)
        if not session or not session.get('mcp_token'):
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='gateway_session',
                    status='failed',
                    event_code='GATEWAY_SESSION_NOT_FOUND',
                    session_credential=gateway_session_id,
                    context={'transport': 'http'},
                )
            raise RuntimeError(DEVICE_INVALID_MESSAGE)
        if diagnostic_emitter is not None:
            diagnostic_emitter.set_defaults(
                session_credential=gateway_session_id,
                admin_id=session.get('admin_id'),
                company_id=session.get('company_id'),
                context={'transport': 'http'},
            )
            diagnostic_emitter.emit(
                stage='gateway_session',
                status='succeeded',
                event_code='GATEWAY_SESSION_RESOLVED',
            )
        return session

    def _enabled_tool_names(self, response):
        if not isinstance(response, dict):
            raise RuntimeError('invalid enabled tool response')
        if response.get('code') != 'MCP_0000':
            raise RuntimeError(response.get('msg') or 'list enabled tools failed')
        data = response.get('data')
        codes = data.get('tool_codes') if isinstance(data, dict) else None
        if not isinstance(codes, list):
            raise RuntimeError('invalid enabled tool response')
        return {
            code.strip().lower()
            for code in codes
            if isinstance(code, str) and code.strip()
        }

    def _load_enabled_tool_names(self, token, request_id=''):
        response = self.api_client.list_enabled_tools(
            token,
            request_id=request_id,
        )
        return self._enabled_tool_names(response)

    def list_tools(self, gateway_session_id, request_id=''):
        session = self._require_session(gateway_session_id)
        request_id = self.build_request_id(request_id)
        enabled = self._load_enabled_tool_names(
            session['mcp_token'],
            request_id,
        )
        return [
            tool.metadata()
            for name, tool in self._tools.items()
            if name in enabled
        ]

    def build_request_id(self, request_id=''):
        request_id = str(request_id or '').strip()
        return request_id or 'rq_{0}'.format(uuid.uuid4().hex[:16])

    def call_tool(
        self,
        gateway_session_id,
        name,
        arguments=None,
        request_id='',
        client_ip='',
        diagnostic_emitter=None,
    ):
        session = self._require_session(gateway_session_id, diagnostic_emitter)
        request_id = self.build_request_id(request_id)
        if name not in self._tools:
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='failed',
                    event_code='TOOL_NOT_REGISTERED',
                    context={'transport': 'http'},
                )
            raise KeyError('tool not registered: {0}'.format(name))

        try:
            enabled_tools = self._load_enabled_tool_names(
                session['mcp_token'],
                request_id,
            )
        except Exception:
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='failed',
                    event_code='ENABLED_TOOL_LOOKUP_FAILED',
                    tool_code=name,
                    context={'transport': 'http'},
                )
            raise
        if name not in enabled_tools:
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='failed',
                    event_code='TOOL_DISABLED',
                    tool_code=name,
                    context={'transport': 'http'},
                )
            raise RuntimeError('tool disabled: {0}'.format(name))

        tool = self._tools[name]

        if diagnostic_emitter is not None:
            diagnostic_emitter.set_defaults(tool_code=name)

        session_hash = hash_gateway_session_id(gateway_session_id)[:12]
        admin_id = session.get('admin_id')
        company_id = session.get('company_id')
        logger.info(
            'MCP public tool call',
            extra={
                'request_id': request_id,
                'tool_code': name,
                'session_hash': session_hash,
                'admin_id': admin_id,
                'company_id': company_id,
            },
        )

        try:
            started_at = time.monotonic()
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='started',
                    event_code='BACKEND_CALL_STARTED',
                )
            result = self.api_client.call_tool(
                token=session['mcp_token'],
                tool_code=tool.name,
                route_path=tool.route_path,
                payload=arguments or {},
                request_id=request_id,
                client_ip=client_ip,
            )
            if hasattr(self.session_store, 'touch_session'):
                self.session_store.touch_session(gateway_session_id)
            logger.info(
                'MCP public tool success',
                extra={
                    'request_id': request_id,
                    'tool_code': name,
                    'session_hash': session_hash,
                    'response_code': result.get('code'),
                },
            )
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='succeeded',
                    event_code='BACKEND_CALL_COMPLETED',
                    response_code=(
                        result.get('code') if isinstance(result, dict) else None
                    ),
                    cost_ms=max(0, int((time.monotonic() - started_at) * 1000)),
                )
            return result
        except Exception as e:
            logger.error(
                'MCP public tool failed',
                extra={
                    'request_id': request_id,
                    'tool_code': name,
                    'session_hash': session_hash,
                    'response_code': 'MCP_9001',
                    'diagnostic_reason': 'UNEXPECTED_EXCEPTION',
                    'exception_class': e.__class__.__name__,
                },
            )
            if diagnostic_emitter is not None:
                diagnostic_emitter.emit(
                    stage='backend_call',
                    status='failed',
                    event_code='UNEXPECTED_EXCEPTION',
                    response_code='MCP_9001',
                )
            raise