"""REST API Client — connect to external APIs and services. Features: - Generic REST client (GET, POST, PUT, DELETE) - Authentication: API key, Bearer token, Basic auth, custom headers - Request/response logging - Rate limiting (configurable requests per second) - Retry with exponential backoff - Timeout handling - JSON and raw response support - Connection pooling (via urllib) Uses only stdlib urllib — no external dependencies. 100% local: no data sent to any cloud service unless explicitly configured. """ from __future__ import annotations import json import logging import time import urllib.error import urllib.request from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass class APIConfig: """Configuration for an API connection.""" name: str base_url: str api_key: str = "" auth_type: str = "api_key" # "api_key", "bearer", "basic", "none", "custom" headers: dict[str, str] = field(default_factory=dict) timeout_s: float = 30.0 max_retries: int = 3 rate_limit_s: float = 0.0 # min seconds between requests retry_backoff: float = 1.5 @dataclass class APIResponse: """Response from an API call.""" success: bool status_code: int data: Any = None error: str = "" elapsed_s: float = 0.0 url: str = "" class RESTClient: """Generic REST API client with auth, retries, and rate limiting.""" def __init__(self, config: APIConfig) -> None: self.config = config self._last_request_time = 0.0 self._stats = { "total_requests": 0, "successful_requests": 0, "failed_requests": 0, "retries": 0, "avg_response_time_s": 0.0, } def _build_headers(self, extra: dict[str, str] | None = None) -> dict[str, str]: """Build request headers with authentication.""" headers = dict(self.config.headers) if extra: headers.update(extra) if self.config.auth_type == "api_key" and self.config.api_key: headers["X-API-Key"] = self.config.api_key elif self.config.auth_type == "bearer" and self.config.api_key: headers["Authorization"] = f"Bearer {self.config.api_key}" elif self.config.auth_type == "basic" and self.config.api_key: import base64 headers["Authorization"] = f"Basic {base64.b64encode(self.config.api_key.encode()).decode()}" return headers def _rate_limit(self) -> None: """Enforce rate limiting.""" if self.config.rate_limit_s > 0: elapsed = time.time() - self._last_request_time if elapsed < self.config.rate_limit_s: time.sleep(self.config.rate_limit_s - elapsed) self._last_request_time = time.time() def request(self, method: str, endpoint: str, data: dict | None = None, params: dict | None = None, headers: dict | None = None) -> APIResponse: """Make an HTTP request. Args: method: GET, POST, PUT, DELETE endpoint: API endpoint (appended to base_url) data: request body (JSON) params: query parameters headers: extra headers """ url = self._build_url(endpoint, params) body = json.dumps(data).encode() if data else None req_headers = self._build_headers(headers) if body: req_headers["Content-Type"] = "application/json" for attempt in range(self.config.max_retries + 1): self._rate_limit() t0 = time.time() self._stats["total_requests"] += 1 try: req = urllib.request.Request(url, data=body, method=method, headers=req_headers) with urllib.request.urlopen(req, timeout=self.config.timeout_s) as resp: raw = resp.read().decode() elapsed = time.time() - t0 self._stats["successful_requests"] += 1 self._update_avg_time(elapsed) try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = raw return APIResponse( success=True, status_code=resp.status, data=parsed, elapsed_s=elapsed, url=url, ) except urllib.error.HTTPError as e: elapsed = time.time() - t0 error_body = "" try: error_body = e.read().decode() except Exception: pass if attempt < self.config.max_retries and e.code >= 500: self._stats["retries"] += 1 wait = self.config.retry_backoff ** (attempt + 1) logger.warning("Retry %d/%d for %s (HTTP %d) after %.1fs", attempt + 1, self.config.max_retries, url, e.code, wait) time.sleep(wait) continue self._stats["failed_requests"] += 1 return APIResponse( success=False, status_code=e.code, error=f"HTTP {e.code}: {error_body[:200]}", elapsed_s=elapsed, url=url, ) except Exception as e: elapsed = time.time() - t0 if attempt < self.config.max_retries: self._stats["retries"] += 1 wait = self.config.retry_backoff ** (attempt + 1) logger.warning("Retry %d/%d for %s: %s", attempt + 1, self.config.max_retries, url, e) time.sleep(wait) continue self._stats["failed_requests"] += 1 return APIResponse( success=False, status_code=0, error=str(e), elapsed_s=elapsed, url=url, ) return APIResponse(success=False, status_code=0, error="Max retries exceeded", url=url) def get(self, endpoint: str, params: dict | None = None) -> APIResponse: return self.request("GET", endpoint, params=params) def post(self, endpoint: str, data: dict | None = None) -> APIResponse: return self.request("POST", endpoint, data=data) def put(self, endpoint: str, data: dict | None = None) -> APIResponse: return self.request("PUT", endpoint, data=data) def delete(self, endpoint: str) -> APIResponse: return self.request("DELETE", endpoint) def _build_url(self, endpoint: str, params: dict | None = None) -> str: url = f"{self.config.base_url.rstrip('/')}/{endpoint.lstrip('/')}" if params: import urllib.parse query = urllib.parse.urlencode(params) url = f"{url}?{query}" return url def _update_avg_time(self, elapsed: float) -> None: total = self._stats["successful_requests"] self._stats["avg_response_time_s"] = ( (self._stats["avg_response_time_s"] * (total - 1) + elapsed) / total ) def get_stats(self) -> dict[str, Any]: return {**self._stats, "name": self.config.name, "base_url": self.config.base_url} class ConnectorRegistry: """Registry of named API connectors. Allows the LLM to connect to multiple external services. Connectors are registered with a name and can be called by the LLM via tool calls: [TOOL: api_call("service_name", "GET", "/endpoint")] """ def __init__(self) -> None: self._connectors: dict[str, RESTClient] = {} self._stats = {"total_connectors": 0, "total_calls": 0} def register(self, name: str, config: APIConfig) -> None: """Register a named API connector.""" config.name = name client = RESTClient(config) self._connectors[name] = client self._stats["total_connectors"] += 1 logger.info("Registered API connector: %s → %s", name, config.base_url) def get(self, name: str) -> RESTClient | None: return self._connectors.get(name) def call(self, name: str, method: str, endpoint: str, data: dict | None = None, params: dict | None = None) -> APIResponse: """Call a registered connector.""" client = self._connectors.get(name) if client is None: return APIResponse(success=False, status_code=0, error=f"Connector '{name}' not found") self._stats["total_calls"] += 1 return client.request(method, endpoint, data=data, params=params) def list_connectors(self) -> list[dict[str, Any]]: return [{"name": name, **client.get_stats()} for name, client in self._connectors.items()] def get_stats(self) -> dict[str, Any]: return {**self._stats, "connectors": {name: c.get_stats() for name, c in self._connectors.items()}}