feat: auth

agent/load_plan_tools.py

@@ -0,0 +1,112 @@
+import os
+from urllib.parse import quote
+
+import httpx
+from dotenv import load_dotenv
+
+from schemas import ShipmentPage, TransportVehicle
+
+load_dotenv()
+
+
+class LoadPlanToolConfigurationError(RuntimeError):
+    pass
+
+
+class LoadPlanToolRequestError(RuntimeError):
+    pass
+
+
+def _required_env(name: str) -> str:
+    value = os.getenv(name, "").strip()
+    if not value:
+        raise LoadPlanToolConfigurationError(f"Missing required environment variable: {name}")
+    return value
+
+
+def _env_positive_float(name: str, default: float) -> float:
+    raw_value = os.getenv(name, str(default)).strip()
+    try:
+        parsed = float(raw_value)
+    except ValueError as exc:
+        raise LoadPlanToolConfigurationError(f"Environment variable {name} must be a number") from exc
+
+    if parsed <= 0:
+        raise LoadPlanToolConfigurationError(f"Environment variable {name} must be greater than 0")
+    return parsed
+
+
+def _api_base_url() -> str:
+    return _required_env("LOAD_PLAN_API_HOST").rstrip("/")
+
+
+def _api_headers() -> dict[str, str]:
+    return {
+        "Authorization": _required_env("LOAD_PLAN_AGENT_ACCESS_KEY"),
+        "Accept": "application/json",
+    }
+
+
+def _build_client() -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        base_url=_api_base_url(),
+        headers=_api_headers(),
+        timeout=_env_positive_float("LOAD_PLAN_API_TIMEOUT_SECONDS", 20.0),
+    )
+
+
+def _raise_for_response(response: httpx.Response) -> None:
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as exc:
+        body = exc.response.text.strip()
+        detail = body or f"status={exc.response.status_code}"
+        raise LoadPlanToolRequestError(
+            f"Load-plan API request failed for {exc.request.method} {exc.request.url}: {detail}"
+        ) from exc
+
+
+async def get_transport_vehicle_by_license_plate(
+    *,
+    merchant_id: int,
+    license_plate: str,
+) -> TransportVehicle:
+    """Fetch a specific transport vehicle for one merchant by license plate."""
+    encoded_plate = quote(license_plate, safe="")
+    path = f"/api/v2/ai/transport-vehicles/{encoded_plate}/"
+
+    async with _build_client() as client:
+        response = await client.get(
+            path,
+            params={"merchant_id": merchant_id},
+        )
+
+    _raise_for_response(response)
+    return TransportVehicle.model_validate(response.json())
+
+
+async def list_unshipped_shipments(
+    *,
+    merchant_id: int,
+    area: str,
+    limit: int | None = None,
+    offset: int | None = None,
+) -> ShipmentPage:
+    """List unshipped shipments for one merchant and one exact area."""
+    params: dict[str, int | str] = {
+        "merchant_id": merchant_id,
+        "area": area,
+    }
+    if limit is not None:
+        params["limit"] = limit
+    if offset is not None:
+        params["offset"] = offset
+
+    async with _build_client() as client:
+        response = await client.get(
+            "/api/v2/ai/shipments/unshipped/",
+            params=params,
+        )
+
+    _raise_for_response(response)
+    return ShipmentPage.model_validate(response.json())