mcp/services/scoped_api_client.py

from services.api_client import JsonTransport


class ScopedApiClient:
    def __init__(self, base_url, transport=None, timeout=10):
        self.base_url = (base_url or '').rstrip('/')
        self.transport = transport or JsonTransport()
        self.timeout = int(timeout)

    def call_tool(self, token, tool_code, route_path, payload, request_id, client_ip=''):
        token = str(token or '').strip()
        if not token:
            raise RuntimeError('mcp token missing')
        url = self.base_url + '/' + route_path.lstrip('/')
        headers = {
            'Authorization': 'Bearer {0}'.format(token),
            'X-MCP-Tool-Code': tool_code,
            'X-Request-Id': request_id,
        }
        client_ip = str(client_ip or '').strip()
        if client_ip:
            headers['X-MCP-Client-IP'] = client_ip
        return self.transport.post_json(url, payload, headers, self.timeout)

    def list_enabled_tools(self, token, request_id=''):
        token = str(token or '').strip()
        if not token:
            raise RuntimeError('mcp token missing')
        url = self.base_url + '/mcp/tools/listEnabledTools'
        headers = {
            'Authorization': 'Bearer {0}'.format(token),
            'X-Request-Id': str(request_id or '').strip(),
        }
        return self.transport.post_json(url, {}, headers, self.timeout)

    def upload_headhaul_document(
        self,
        token,
        upload_token,
        filename,
        file_bytes,
        content_type='',
        request_id='',
        client_ip='',
    ):
        token = str(token or '').strip()
        if not token:
            raise RuntimeError('mcp token missing')
        url = self.base_url + '/mcp/tools/uploadHeadhaulDocument'
        headers = {
            'Authorization': 'Bearer {0}'.format(token),
            'X-MCP-Tool-Code': 'prepare_headhaul_document_upload',
            'X-Request-Id': str(request_id or '').strip(),
        }
        client_ip = str(client_ip or '').strip()
        if client_ip:
            headers['X-MCP-Client-IP'] = client_ip
        return self.transport.post_multipart(
            url,
            {'upload_token': upload_token},
            {
                'file': (
                    filename,
                    file_bytes,
                    content_type or 'application/octet-stream',
                ),
            },
            headers,
            max(self.timeout, 120),
        )