client.py•3.24 kB
"""HTTP client for interacting with the ALA API."""
import logging
from typing import Any
import httpx
from .config import settings
logger = logging.getLogger("ala_mcp.client")
class ALAClient:
"""HTTP client for making requests to the ALA API."""
def __init__(
self,
base_url: str | None = None,
timeout: int | None = None,
api_key: str | None = None,
):
"""
Initialize the ALA API client.
Args:
base_url: Base URL for the API (defaults to settings)
timeout: Request timeout in seconds (defaults to settings)
api_key: Optional API key for authentication
"""
self.base_url = (base_url or settings.base_url).rstrip("/")
self.timeout = timeout or settings.default_timeout
self.api_key = api_key or settings.api_key
async def request(
self,
method: str,
endpoint: str,
params: dict[str, Any] | None = None,
data: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
) -> dict[str, Any]:
"""
Make an HTTP request to the ALA API.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint path
params: Query parameters
data: Request body data
headers: Additional headers
Returns:
Dict containing status_code, headers, url, and data
Raises:
httpx.HTTPError: If the request fails
"""
# Construct full URL
url = f"{self.base_url}/{endpoint.lstrip('/')}"
# Set default headers
default_headers = {"Accept": "application/json"}
if self.api_key:
default_headers["Authorization"] = f"Bearer {self.api_key}"
if headers:
default_headers.update(headers)
logger.info(f"Making {method} request to {url}")
logger.debug(f"Params: {params}, Data: {data}")
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.request(
method=method,
url=url,
params=params,
json=data if data else None,
headers=default_headers,
)
# Log response
logger.info(f"Response status: {response.status_code}")
# Format response
result = {
"status_code": response.status_code,
"headers": dict(response.headers),
"url": str(response.url),
}
# Try to parse JSON, fallback to text
try:
result["data"] = response.json()
except Exception:
result["data"] = response.text
return result
except httpx.TimeoutException as e:
logger.error(f"Request timeout: {e}")
raise
except httpx.HTTPError as e:
logger.error(f"HTTP error: {e}")
raise
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise