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_export_task import QueryExportTaskTool 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), 'query_export_task': QueryExportTaskTool(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