mcp/app.py

import argparse
import json
import sys
import uuid

from config import GatewayConfig
from constants import DEVICE_INVALID_MESSAGE
from mcp_protocol import McpProtocolHandler
from public_gateway import PublicGatewayApp
from public_server import serve_public
from services.api_client import ApiClient
from services.auth_client import AuthClient
from services.gateway_session_store import GatewaySessionStore
from services.diagnostic_reporter import (
    NullDiagnosticReporter,
    diagnostic_reporter_from_config,
)
from services.scoped_api_client import ScopedApiClient
from services.token_store import FileTokenStore, RedisSocketClient, RedisTokenStore
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


def parse_int_list(value):
    if not value:
        return []
    return [int(item.strip()) for item in value.split(',') if item.strip()]


def parse_string_list(value):
    result = []
    for item in (value or '').split(','):
        item = item.strip()
        if item and item not in result:
            result.append(item)
    return result


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

    @classmethod
    def from_config(cls, config, redis_client=None):
        if config.token_store_type == 'redis':
            redis = redis_client or RedisSocketClient(
                host=config.redis_host,
                port=config.redis_port,
                db=config.redis_db,
                password=config.redis_password,
                timeout=config.timeout_seconds,
            )
            token_store = RedisTokenStore(
                redis,
                prefix=config.redis_prefix,
                session_key=config.session_key,
                refresh_skew_seconds=config.refresh_skew_seconds,
            )
        elif config.token_store_type == 'file':
            token_store = FileTokenStore(
                config.token_store_path,
                refresh_skew_seconds=config.refresh_skew_seconds,
            )
        else:
            raise ValueError('unsupported token store type: {0}'.format(config.token_store_type))

        auth_client = AuthClient(
            base_url=config.auth_base_url,
            client_type=config.client_type,
            token_store=token_store,
            timeout=config.timeout_seconds,
            session_key=config.session_key,
        )
        api_client = ApiClient(
            base_url=config.tools_base_url,
            token_store=token_store,
            timeout=config.timeout_seconds,
        )
        return cls(
            auth_client=auth_client,
            api_client=api_client,
            token_store=token_store,
            reporter=diagnostic_reporter_from_config(config),
        )

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

    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, request_id=''):
        if self.api_client is None or not hasattr(self.api_client, 'list_enabled_tools'):
            raise RuntimeError('enabled tool client unavailable')
        return self._enabled_tool_names(
            self.api_client.list_enabled_tools(request_id=request_id)
        )

    def list_tools(self, request_id=''):
        request_id = self.build_request_id(request_id)
        enabled = self._load_enabled_tool_names(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()
        if request_id:
            return request_id
        return 'rq_{0}'.format(uuid.uuid4().hex[:16])

    def ensure_session(self):
        if self.token_store is None:
            return
        session = self.token_store.get()
        if not session or not session.get('token'):
            raise RuntimeError(DEVICE_INVALID_MESSAGE)
        if self.token_store.is_expiring():
            if self.auth_client is None:
                raise RuntimeError('mcp token expiring but auth client missing')
            self.auth_client.refresh(session['token'])

    def call_tool(self, name, arguments=None, request_id=''):
        if name not in self._tools:
            raise KeyError('tool not registered: {0}'.format(name))
        tool = self._tools[name]
        if getattr(tool, 'requires_session', True):
            self.ensure_session()
        request_id = self.build_request_id(request_id)
        if name not in self._load_enabled_tool_names(request_id):
            raise RuntimeError('tool disabled: {0}'.format(name))
        arguments = arguments or {}
        return tool.call(request_id=request_id, **arguments)

    def create_protocol_handler(self):
        return McpProtocolHandler(self, reporter=self.reporter)

    def run_cli(self, argv=None, stdin=None, stdout=None):
        stdin = stdin or sys.stdin
        stdout = stdout or sys.stdout
        parser = argparse.ArgumentParser(prog='mcp-gateway')
        subparsers = parser.add_subparsers(dest='command', required=True)

        subparsers.add_parser('list-tools')
        subparsers.add_parser('serve-stdio')

        public_parser = subparsers.add_parser('serve-public')
        public_parser.add_argument('--host', default='0.0.0.0')
        public_parser.add_argument('--port', type=int, default=8765)

        call_parser = subparsers.add_parser('call')
        call_parser.add_argument('--tool', required=True)
        call_parser.add_argument('--keyword', default='')
        call_parser.add_argument('--order-id', type=int, default=0)
        call_parser.add_argument('--order-number', default='')
        call_parser.add_argument('--section', default='全部')
        call_parser.add_argument('--order-numbers', default='')
        call_parser.add_argument('--tracking-number', default='')
        call_parser.add_argument('--tracking-numbers', default='')
        call_parser.add_argument('--reference-number', default='')
        call_parser.add_argument('--reference-numbers', default='')
        call_parser.add_argument('--outbound-number', default='')
        call_parser.add_argument('--outbound-numbers', default='')
        call_parser.add_argument('--bl-numbers', default='')
        call_parser.add_argument('--container-code', default='')
        call_parser.add_argument('--container-codes', default='')
        call_parser.add_argument('--so-number', default='')
        call_parser.add_argument('--so-numbers', default='')
        call_parser.add_argument('--shipment-id', default='')
        call_parser.add_argument('--receiver-country', default='')
        call_parser.add_argument('--product-ids', default='')
        call_parser.add_argument('--customer-ids', default='')
        call_parser.add_argument('--sales-id', type=int, default=0)
        call_parser.add_argument('--warehouse-ids', default='')
        call_parser.add_argument('--warehouse-id', type=int, default=0)
        call_parser.add_argument('--department-id', type=int, default=0)
        call_parser.add_argument('--outbound-status', type=int, default=0)
        call_parser.add_argument('--shipping-method', type=int, default=0)
        call_parser.add_argument('--is-direct-send', type=int, default=None)
        call_parser.add_argument('--trailer-types', default='')
        call_parser.add_argument('--declaration-types', default='')
        call_parser.add_argument('--clearance-types', default='')
        call_parser.add_argument('--closing-time-start', default='')
        call_parser.add_argument('--closing-time-end', default='')
        call_parser.add_argument('--est-loading-time-start', default='')
        call_parser.add_argument('--est-loading-time-end', default='')
        call_parser.add_argument('--create-date-start', default='')
        call_parser.add_argument('--create-date-end', default='')
        call_parser.add_argument('--loading-time-start', default='')
        call_parser.add_argument('--loading-time-end', default='')
        call_parser.add_argument('--inbound-date-start', default='')
        call_parser.add_argument('--inbound-date-end', default='')
        call_parser.add_argument('--outbound-date-start', default='')
        call_parser.add_argument('--outbound-date-end', default='')
        call_parser.add_argument('--filter-type', default='')
        call_parser.add_argument('--task-ref', default='')
        call_parser.add_argument('--page', type=int, default=1)
        call_parser.add_argument('--limit', type=int, default=20)
        call_parser.add_argument('--request-id', default='')

        args = parser.parse_args(argv or [])

        if args.command == 'list-tools':
            payload = self.list_tools()
        elif args.command == 'serve-stdio':
            try:
                return self.create_protocol_handler().run_stdio(
                    stdin=stdin,
                    stdout=stdout,
                )
            finally:
                self.reporter.close()
        elif args.command == 'serve-public':
            config = GatewayConfig.from_env()
            reporter = self.reporter
            if isinstance(reporter, NullDiagnosticReporter):
                reporter = diagnostic_reporter_from_config(config)
            redis = RedisSocketClient(
                host=config.redis_host,
                port=config.redis_port,
                db=config.redis_db,
                password=config.redis_password,
                timeout=config.timeout_seconds,
            )
            session_store = GatewaySessionStore(
                redis,
                prefix=config.redis_prefix,
                ttl_seconds=config.gateway_session_ttl_seconds,
            )
            public_app = PublicGatewayApp(
                session_store=session_store,
                api_client=ScopedApiClient(config.tools_base_url, timeout=config.timeout_seconds),
            )
            try:
                return serve_public(
                    public_app,
                    host=args.host,
                    port=args.port,
                    enable_rate_limit=config.rate_limit_enabled,
                    rate_limit_max_requests=config.rate_limit_max_requests,
                    rate_limit_window_seconds=config.rate_limit_window_seconds,
                    max_in_flight_per_tool=config.max_in_flight_per_tool,
                    reporter=reporter,
                )
            finally:
                reporter.close()
        elif args.command == 'call':
            tool_args = {
                'page': args.page,
                'limit': args.limit,
            }

            if args.tool == 'query_order':
                if not args.keyword:
                    raise ValueError('--keyword is required for query_order')
                tool_args['keyword'] = args.keyword
            elif args.tool == 'query_track':
                if args.order_id > 0:
                    tool_args['order_id'] = args.order_id
                if args.order_number:
                    tool_args['order_number'] = args.order_number
                if args.tracking_number:
                    tool_args['tracking_number'] = args.tracking_number
                if args.order_id <= 0 and not args.order_number and not args.tracking_number:
                    raise ValueError('--order-id, --order-number or --tracking-number is required for query_track')
            elif args.tool == 'query_order_exact':
                exact_strings = {
                    'order_number': args.order_number,
                    'reference_number': args.reference_number,
                    'tracking_number': args.tracking_number,
                    'outbound_number': args.outbound_number,
                    'container_code': args.container_code,
                    'so_number': args.so_number,
                    'shipment_id': args.shipment_id,
                    'receiver_country': args.receiver_country,
                    'inbound_date_start': args.inbound_date_start,
                    'inbound_date_end': args.inbound_date_end,
                    'outbound_date_start': args.outbound_date_start,
                    'outbound_date_end': args.outbound_date_end,
                }
                for field, value in exact_strings.items():
                    if value:
                        tool_args[field] = value
                exact_number_lists = {
                    'order_numbers': args.order_numbers,
                    'reference_numbers': args.reference_numbers,
                    'tracking_numbers': args.tracking_numbers,
                    'outbound_numbers': args.outbound_numbers,
                    'container_codes': args.container_codes,
                    'so_numbers': args.so_numbers,
                }
                for field, value in exact_number_lists.items():
                    if value:
                        tool_args[field] = parse_string_list(value)
                exact_lists = {
                    'product_ids': args.product_ids,
                    'customer_ids': args.customer_ids,
                    'warehouse_ids': args.warehouse_ids,
                }
                for field, value in exact_lists.items():
                    if value:
                        tool_args[field] = parse_int_list(value)
                if args.sales_id > 0:
                    tool_args['sales_id'] = args.sales_id
                if args.department_id > 0:
                    tool_args['department_id'] = args.department_id
            elif args.tool == 'query_order_detail':
                if not args.order_number:
                    raise ValueError('--order-number is required for query_order_detail')
                tool_args['order_number'] = args.order_number
                tool_args['section'] = args.section
            elif args.tool == 'query_customs_declaration_files':
                if args.outbound_numbers:
                    tool_args['outbound_numbers'] = parse_string_list(
                        args.outbound_numbers
                    )
                if args.order_numbers:
                    tool_args['order_numbers'] = parse_string_list(
                        args.order_numbers
                    )
            elif args.tool == 'query_outbound_list':
                outbound_number_lists = {
                    'outbound_numbers': args.outbound_numbers,
                    'order_numbers': args.order_numbers,
                    'container_codes': args.container_codes,
                    'so_numbers': args.so_numbers,
                    'bl_numbers': args.bl_numbers,
                }
                for field, value in outbound_number_lists.items():
                    if value:
                        tool_args[field] = parse_string_list(value)
                outbound_mode_lists = {
                    'trailer_types': args.trailer_types,
                    'declaration_types': args.declaration_types,
                    'clearance_types': args.clearance_types,
                }
                for field, value in outbound_mode_lists.items():
                    if value:
                        tool_args[field] = parse_int_list(value)
                outbound_dates = {
                    'closing_time_start': args.closing_time_start,
                    'closing_time_end': args.closing_time_end,
                    'est_loading_time_start': args.est_loading_time_start,
                    'est_loading_time_end': args.est_loading_time_end,
                    'create_date_start': args.create_date_start,
                    'create_date_end': args.create_date_end,
                    'loading_time_start': args.loading_time_start,
                    'loading_time_end': args.loading_time_end,
                }
                for field, value in outbound_dates.items():
                    if value:
                        tool_args[field] = value
                if args.outbound_status > 0:
                    tool_args['outbound_status'] = args.outbound_status
                if args.shipping_method > 0:
                    tool_args['shipping_method'] = args.shipping_method
                if args.warehouse_id == -1 or args.warehouse_id > 0:
                    tool_args['warehouse_id'] = args.warehouse_id
                if args.is_direct_send is not None:
                    tool_args['is_direct_send'] = args.is_direct_send
            elif args.tool == 'query_outbound_detail':
                if not args.outbound_number:
                    raise ValueError(
                        '--outbound-number is required for '
                        'query_outbound_detail'
                    )
                tool_args['outbound_number'] = args.outbound_number
            elif args.tool == 'query_export_task':
                if not args.task_ref:
                    raise ValueError(
                        '--task-ref is required for query_export_task'
                    )
                tool_args = {'task_ref': args.task_ref}
            elif args.tool in (
                'list_order_filter_options', 'list_outbound_filter_options'
            ):
                if not args.filter_type:
                    raise ValueError(
                        '--filter-type is required for '
                        + args.tool
                    )
                tool_args['filter_type'] = args.filter_type
                tool_args['keyword'] = args.keyword
            else:
                if args.keyword:
                    tool_args['keyword'] = args.keyword

            payload = self.call_tool(
                args.tool,
                tool_args,
                request_id=args.request_id,
            )
        else:
            raise RuntimeError('unsupported command')

        stdout.write(json.dumps(payload, ensure_ascii=False))
        return 0


def main(argv=None):
    config = GatewayConfig.from_env()
    app = GatewayApp.from_config(config)
    return app.run_cli(argv=argv)


if __name__ == '__main__':
    raise SystemExit(main(sys.argv[1:]))