mcp/tools/query_customer_payment_records.py

import re
from datetime import date


class QueryCustomerPaymentRecordsTool:
    name = 'query_customer_payment_records'
    route_path = '/mcp/tools/queryCustomerPaymentRecords'

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

    def metadata(self):
        return {
            'name': self.name,
            'description': (
                '分页查询单个客户的逐笔收款记录、已核销金额和未核销金额。必须先调用'
                'list_customer_filter_options并使用客户名称筛选取得customer_id；不得根据名称猜测'
                'customer_id，不得模糊查询或改用其他工具试查。收款日期可按闭区间筛选，'
                '起止日期同时提供时闭区间最多包含366个日历日。公司、员工、菜单权限和客户数据范围由'
                '当前设备会话确定，调用方不得覆盖。'
            ),
            'input_schema': {
                'type': 'object',
                'properties': {
                    'customer_id': {'type': 'integer', 'minimum': 1},
                    'receive_date_start': {
                        'type': 'string', 'format': 'date',
                        'pattern': '^\\d{4}-\\d{2}-\\d{2}$',
                    },
                    'receive_date_end': {
                        'type': 'string', 'format': 'date',
                        'pattern': '^\\d{4}-\\d{2}-\\d{2}$',
                    },
                    'page': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 1,
                    },
                    'limit': {
                        'type': 'integer', 'minimum': 1, 'maximum': 100,
                        'default': 20,
                    },
                },
                'required': ['customer_id'],
                'additionalProperties': False,
            },
        }

    def call(
        self,
        customer_id,
        receive_date_start=None,
        receive_date_end=None,
        page=1,
        limit=20,
        request_id='rq_query_customer_payment_records',
    ):
        if self.api_client is None:
            raise RuntimeError(
                'api client is required for query_customer_payment_records'
            )
        start = self._optional_date(receive_date_start, 'receive_date_start')
        end = self._optional_date(receive_date_end, 'receive_date_end')
        if start and end:
            span = (date.fromisoformat(end) - date.fromisoformat(start)).days
            if span < 0 or span > 365:
                raise ValueError('receive date range is invalid')
        payload = {
            'customer_id': self._positive_integer(customer_id, 'customer_id'),
            'page': self._bounded_integer(page, 'page'),
            'limit': self._bounded_integer(limit, 'limit'),
        }
        if start:
            payload['receive_date_start'] = start
        if end:
            payload['receive_date_end'] = end
        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

    @staticmethod
    def _optional_date(value, field):
        if value is None or value == '':
            return ''
        if not isinstance(value, str):
            raise ValueError('{0} is invalid'.format(field))
        if re.fullmatch(r'[0-9]{4}-[0-9]{2}-[0-9]{2}', value) is None:
            raise ValueError('{0} is invalid'.format(field))
        try:
            date.fromisoformat(value)
        except ValueError:
            raise ValueError('{0} is invalid'.format(field))
        return value