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)