compass_backoffice/mcp_servers/backoffice_mcp_server/clients.py

from __future__ import annotations

from typing import Any, Dict

import httpx

from .settings import BackofficeMCPSettings


class BackofficeRESTClient:
    def __init__(self, settings: BackofficeMCPSettings):
        self.settings = settings
        self.base_url = settings.rest_base_url.rstrip("/")

    def _headers(self) -> Dict[str, str]:
        if not self.settings.rest_api_key:
            return {}
        value = self.settings.rest_api_key
        if self.settings.rest_auth_header.lower() == "authorization" and not value.lower().startswith("bearer "):
            value = f"Bearer {value}"
        return {self.settings.rest_auth_header: value}

    def get_json(self, path: str, params: Dict[str, Any]) -> Dict[str, Any]:
        if not self.base_url:
            raise RuntimeError("BACKOFFICE_MCP_REST_BASE_URL não configurado")
        with httpx.Client(timeout=self.settings.rest_timeout_seconds, headers=self._headers()) as client:
            resp = client.get(f"{self.base_url}/{path.lstrip('/')}", params=params)
            resp.raise_for_status()
            return resp.json()

    def post_json(self, path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        if not self.base_url:
            raise RuntimeError("BACKOFFICE_MCP_REST_BASE_URL não configurado")
        with httpx.Client(timeout=self.settings.rest_timeout_seconds, headers=self._headers()) as client:
            resp = client.post(f"{self.base_url}/{path.lstrip('/')}", json=payload)
            resp.raise_for_status()
            return resp.json()