mcp/public_gateway.py

import logging
import uuid

from constants import DEVICE_INVALID_MESSAGE
from tools.list_order_filter_options import ListOrderFilterOptionsTool
from tools.query_order import QueryOrderTool
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),
            'list_order_filter_options': ListOrderFilterOptionsTool(
                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):
        response = self.api_client.list_enabled_tools(token)
        return self._enabled_tool_names(response)

    def list_tools(self, gateway_session_id):
        session = self._require_session(gateway_session_id)
        enabled = self._load_enabled_tool_names(session['mcp_token'])
        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)
        if name not in self._load_enabled_tool_names(session['mcp_token']):
            raise RuntimeError('tool disabled: {0}'.format(name))

        tool = self._tools[name]
        request_id = self.build_request_id(request_id)

        session_hash = hash_gateway_session_id(gateway_session_id)[:12]
        admin_id = session.get('admin_id')
        company_id = session.get('company_id')
        logger.info(f"[AUDIT] tool_call: session_hash={session_hash}, admin_id={admin_id}, company_id={company_id}, tool={name}, request_id={request_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(f"[AUDIT] tool_success: session_hash={session_hash}, tool={name}, request_id={request_id}, code={result.get('code')}")
            return result
        except Exception as e:
            logger.error(f"[AUDIT] tool_error: session_hash={session_hash}, tool={name}, request_id={request_id}, error={str(e)}")
            raise