mcp/services/auth_client.py

from services.api_client import JsonTransport


class AuthClient:
    def __init__(self, base_url, client_type, token_store, transport=None, timeout=10, session_key=''):
        self.base_url = (base_url or '').rstrip('/')
        self.client_type = client_type
        self.token_store = token_store
        self.transport = transport or JsonTransport()
        self.timeout = int(timeout)
        self.session_key = str(session_key or '').strip()

    def exchange(self, auth_code):
        payload = {'auth_code': auth_code, 'client_type': self.client_type}
        if self.session_key:
            payload['session_key'] = self.session_key
        response = self.transport.post_json(
            self.base_url + '/admin/mcp/auth/exchange',
            payload,
            {},
            self.timeout,
        )
        self._persist_token(response)
        return response

    def refresh(self, mcp_token):
        response = self.transport.post_json(
            self.base_url + '/admin/mcp/auth/refresh',
            {'mcp_token': mcp_token},
            {'Authorization': 'Bearer {0}'.format(mcp_token)},
            self.timeout,
        )
        self._persist_token(response)
        return response

    def revoke(self, mcp_token):
        response = self.transport.post_json(
            self.base_url + '/admin/mcp/auth/revoke',
            {'mcp_token': mcp_token},
            {'Authorization': 'Bearer {0}'.format(mcp_token)},
            self.timeout,
        )
        if (response.get('code') or '') == 'MCP_0000':
            self.token_store.clear()
        return response

    def _persist_token(self, response):
        data = response.get('data') or {}
        token = data.get('mcp_token')
        expire_time = data.get('expire_time')
        if token and expire_time:
            self.token_store.save(token, expire_time)