mcp/public_gateway.py

import logging
import uuid

from constants import BIND_HINT
from tools.bind_auth_code import BindAuthCodeTool
from tools.query_order import QueryOrderTool
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):
        self.session_store = session_store
        self.api_client = api_client
        self.auth_client = auth_client
        self._tools = {
            'bind_auth_code': BindAuthCodeTool(auth_client=None),
            'query_order': QueryOrderTool(api_client=None),
            'query_track': QueryTrackTool(api_client=None),
        }

    def list_tools(self):
        return [tool.metadata() for tool in self._tools.values()]

    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 bind_auth_code(self, gateway_session_id, auth_code):
        if self.auth_client is None:
            raise RuntimeError('auth client is required')

        session_hash = hash_gateway_session_id(gateway_session_id)[:12]
        logger.info(f"[AUDIT] bind_auth_code: session_hash={session_hash}, auth_code_provided={bool(auth_code)}")

        result = self.auth_client.exchange(gateway_session_id, auth_code)

        if result.get('code') == 'MCP_0000':
            session = self.session_store.get(gateway_session_id)
            admin_id = session.get('admin_id') if session else None
            company_id = session.get('company_id') if session else None
            logger.info(f"[AUDIT] bind_success: session_hash={session_hash}, admin_id={admin_id}, company_id={company_id}")
        else:
            logger.warning(f"[AUDIT] bind_failed: session_hash={session_hash}, code={result.get('code')}, msg={result.get('msg')}")

        return result

    def call_tool(self, gateway_session_id, name, arguments=None, request_id=''):
        if name == 'bind_auth_code':
            auth_code = (arguments or {}).get('auth_code', '')
            return self.bind_auth_code(gateway_session_id, auth_code)
        if name not in self._tools:
            raise KeyError('tool not registered: {0}'.format(name))

        session = self.session_store.get(gateway_session_id)
        if not session or not session.get('mcp_token'):
            raise RuntimeError(BIND_HINT)

        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,
            )
            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