mcp/tools/query_customer_list.py

class QueryCustomerListTool:
    name = 'query_customer_list'
    route_path = '/mcp/tools/queryCustomerList'

    def __init__(self, api_client=None):
        self.api_client = api_client

    def metadata(self):
        positive_id = {'type': 'integer', 'minimum': 1}
        return {
            'name': self.name,
            'description': (
                '查询当前员工有权查看的启用主客户列表。可按客户名称、事业部、商务经理或'
                '客户经理筛选；筛选值必须先调用list_customer_filter_options取得，禁止猜测'
                '内部ID。未指定筛选时查询全部授权客户。公司、员工和权限范围由当前设备会话'
                '确定，调用方不得覆盖。结算模式按业务类型返回结构化列表，展示时必须完整保留'
                '每个业务类型和对应结算模式，不得合并或缩写。最终回复必须完整逐条展示17个字段，'
                '包括空字段，不得摘要、隐藏、重命名或省略任何字段。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'customer_id': dict(positive_id),
                    'department_id': dict(positive_id),
                    'sales_id': dict(positive_id),
                    'merchandiser_id': dict(positive_id),
                    'page': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 20,
                    },
                },
                'required': [],
                'additionalProperties': False,
            },
        }

    def call(
        self, customer_id=None, department_id=None, sales_id=None,
        merchandiser_id=None, page=1, limit=20,
        request_id='rq_query_customer_list',
    ):
        if self.api_client is None:
            raise RuntimeError('api client is required for query_customer_list')
        payload = {
            'page': self._bounded_integer(page, 'page'),
            'limit': self._bounded_integer(limit, 'limit'),
        }
        for field, value in (
            ('customer_id', customer_id),
            ('department_id', department_id),
            ('sales_id', sales_id),
            ('merchandiser_id', merchandiser_id),
        ):
            if value is not None:
                payload[field] = self._positive_integer(value, field)
        return self.api_client.call_tool(
            self.name, self.route_path, payload, request_id
        )

    @staticmethod
    def _positive_integer(value, field):
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError('{0} is invalid'.format(field))
        return value

    @classmethod
    def _bounded_integer(cls, value, field):
        value = cls._positive_integer(value, field)
        if value > 100:
            raise ValueError('{0} is invalid'.format(field))
        return value