mcp/services/api_client.py

import json
import urllib.request


class JsonTransport:
    def post_json(self, url, payload, headers, timeout):
        request = urllib.request.Request(
            url=url,
            data=json.dumps(payload).encode('utf-8'),
            headers=dict({'Content-Type': 'application/json'}, **headers),
            method='POST',
        )
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return json.loads(response.read().decode('utf-8-sig'))


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

    def call_tool(self, tool_code, route_path, payload, request_id):
        token = self.token_store.require_token()
        url = self.base_url + '/' + route_path.lstrip('/')
        headers = {
            'Authorization': 'Bearer {0}'.format(token),
            'X-MCP-Tool-Code': tool_code,
            'X-Request-Id': request_id,
        }
        return self.transport.post_json(url, payload, headers, self.timeout)

    def list_enabled_tools(self, request_id=''):
        token = self.token_store.require_token()
        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)