mcp/public_gateway.py

import logging
import uuid

from constants import DEVICE_INVALID_MESSAGE
from tools.list_order_filter_options import ListOrderFilterOptionsTool
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_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_customs_declaration_files':
                QueryCustomsDeclarationFilesTool(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):
        session = self.session_store.get(gateway_session_id)
        if not session or not session.get('mcp_token'):
            raise RuntimeError(DEVICE_INVALID_MESSAGE)
        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=''):
        if name not in self._tools:
            raise KeyError('tool not registered: {0}'.format(name))

        session = self._require_session(gateway_session_id)
        request_id = self.build_request_id(request_id)
        if name not in self._load_enabled_tool_names(
            session['mcp_token'],
            request_id,
        ):
            raise RuntimeError('tool disabled: {0}'.format(name))

        tool = self._tools[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:
            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'),
                },
            )
            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__,
                },
            )
            raise