"""
API Client for Zintlr MCP Server.
This module replicates the Next.js proxy logic from server-request-handler.ts,
allowing the MCP server to call api.zintlr.com directly (bypassing auth.zintlr.com).
Key responsibilities:
1. Decrypt JWT tokens using verify_and_decrypt_jwt
2. Build proper headers (Authorization, visitor-id, client-ip-address, Captcha-Token)
3. Add decrypted key to request body
4. Make HTTP requests to api.zintlr.com
"""
import httpx
from typing import Any
from app.config import settings
from app.crypto import verify_and_decrypt_jwt
class ZintlrAPIClient:
"""HTTP client for calling Zintlr Django APIs directly."""
def __init__(self, user_tokens: dict[str, str], client_ip: str = "Unknown"):
"""
Initialize the API client with user tokens.
Args:
user_tokens: Dict containing encrypted tokens from Zintlr login:
- access_token: Encrypted JWT access token
- key: Encrypted user key/hash
- visitor_id: Visitor ID (not encrypted)
client_ip: Client's IP address for tracking
"""
self.encrypted_access_token = user_tokens.get("access_token", "")
self.encrypted_key = user_tokens.get("key", "")
self.visitor_id = user_tokens.get("visitor_id", "")
self.client_ip = client_ip
# Decrypt tokens
self._decrypted_access_token = None
self._decrypted_key = None
def _ensure_decrypted(self) -> None:
"""Decrypt tokens if not already done."""
if self._decrypted_access_token is None:
self._decrypted_access_token = verify_and_decrypt_jwt(self.encrypted_access_token)
if self._decrypted_access_token is None:
raise ValueError("Failed to decrypt access_token - token may be invalid or expired")
if self._decrypted_key is None:
self._decrypted_key = verify_and_decrypt_jwt(self.encrypted_key)
if self._decrypted_key is None:
raise ValueError("Failed to decrypt key - token may be invalid or expired")
def _build_headers(self) -> dict[str, str]:
"""
Build request headers replicating Next.js proxy logic.
From server-request-handler.ts:
- headers["Authorization"] = decrypted_access_token
- headers["visitor-id"] = visitor_id
- headers["client-ip-address"] = ip_address
- headers["Captcha-Token"] = CAPTCHA_TOKEN
"""
self._ensure_decrypted()
return {
"Authorization": str(self._decrypted_access_token),
"visitor-id": self.visitor_id,
"client-ip-address": self.client_ip,
"Captcha-Token": settings.captcha_token,
"Content-Type": "application/json",
"Accept": "application/json",
}
def _build_body(self, data: dict[str, Any]) -> dict[str, Any]:
"""
Build request body with decrypted key added.
From server-request-handler.ts:
- data.key = decrypted_key
- data.ip_address = ip_address
"""
self._ensure_decrypted()
body = {**data}
body["key"] = self._decrypted_key
body["ip_address"] = self.client_ip
return body
async def request(
self,
method: str,
endpoint: str,
data: dict[str, Any] | None = None,
timeout: float = 60.0,
) -> dict[str, Any]:
"""
Make an API request to api.zintlr.com.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint path (e.g., "prospecting/search")
data: Request data/parameters
timeout: Request timeout in seconds
Returns:
API response as dict
"""
data = data or {}
url = f"{settings.zintlr_api_base_url}/{endpoint.lstrip('/')}"
headers = self._build_headers()
async with httpx.AsyncClient(timeout=timeout) as client:
if method.upper() == "GET":
response = await client.get(url, params=data, headers=headers)
else:
body = self._build_body(data)
response = await client.request(
method=method.upper(),
url=url,
json=body,
headers=headers,
)
try:
return response.json()
except Exception:
return {
"success": False,
"message": f"Invalid response from server (status: {response.status_code})",
"code": response.status_code,
}
async def get(self, endpoint: str, params: dict[str, Any] | None = None, **kwargs) -> dict[str, Any]:
"""Make a GET request."""
return await self.request("GET", endpoint, params, **kwargs)
async def post(self, endpoint: str, data: dict[str, Any] | None = None, **kwargs) -> dict[str, Any]:
"""Make a POST request."""
return await self.request("POST", endpoint, data, **kwargs)
async def make_api_request(
endpoint: str,
method: str,
data: dict[str, Any],
user_tokens: dict[str, str],
client_ip: str = "Unknown",
) -> tuple[bool, dict[str, Any] | None, str | None]:
"""
Convenience function to make an API request.
Args:
endpoint: API endpoint path
method: HTTP method
data: Request data
user_tokens: User's encrypted tokens
client_ip: Client IP address
Returns:
Tuple of (success: bool, data: dict | None, error_message: str | None)
"""
try:
client = ZintlrAPIClient(user_tokens, client_ip)
response = await client.request(method, endpoint, data)
if response.get("success"):
return True, response, None
else:
return False, response, response.get("message", "Unknown error")
except ValueError as e:
return False, None, str(e)
except httpx.TimeoutException:
return False, None, "Request timed out"
except httpx.RequestError as e:
return False, None, f"Request failed: {str(e)}"
except Exception as e:
return False, None, f"Unexpected error: {str(e)}"
