App/mcp/api_client.py

from __future__ import annotations

import os
from typing import Any, Optional

import httpx


DEFAULT_API_BASE_URL = "https://mbonea-investingtest.hf.space"


def get_api_base_url() -> Optional[str]:
    """Return remote API base URL, or None when MCP should use the local database."""
    if os.getenv("UWEKEZAJI_MCP_USE_LOCAL", "").strip().lower() in {"1", "true", "yes"}:
        return None
    return os.getenv("UWEKEZAJI_API_BASE_URL", DEFAULT_API_BASE_URL).rstrip("/")


def days_to_time_range(days: int) -> str:
    days = max(1, min(days, 3650))
    if days <= 7:
        return "1w"
    if days <= 30:
        return "1m"
    if days <= 180:
        return "6m"
    if days <= 365:
        return "1y"
    if days <= 730:
        return "2y"
    return "5y"


class UwekezajiApiClient:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")
        self._client = httpx.AsyncClient(timeout=45.0, follow_redirects=True)

    async def close(self) -> None:
        await self._client.aclose()

    async def _get(self, path: str, params: Optional[dict[str, Any]] = None) -> dict[str, Any]:
        response = await self._client.get(f"{self.base_url}{path}", params=params)
        response.raise_for_status()
        payload = response.json()
        if not payload.get("success"):
            raise RuntimeError(payload.get("message") or f"API request failed for {path}")
        data = payload.get("data")
        return data if isinstance(data, dict) else {}

    async def list_stocks(self) -> list[dict[str, Any]]:
        data = await self._get("/stocks/list")
        stocks = data.get("stocks") or []
        return stocks if isinstance(stocks, list) else []

    async def get_prices(
        self,
        symbol: str,
        *,
        time_range: str = "6m",
        limit: int = 500,
    ) -> list[dict[str, Any]]:
        data = await self._get(
            f"/stocks/{symbol.upper()}/prices",
            params={"time_range": time_range, "limit": limit},
        )
        prices = data.get("prices") or []
        return prices if isinstance(prices, list) else []

    async def get_fundamentals(self, symbol: str) -> dict[str, Any]:
        data = await self._get(f"/stocks/{symbol.upper()}/fundamentals")
        fundamentals = data.get("fundamentals") or {}
        return fundamentals if isinstance(fundamentals, dict) else {}

    async def get_company(self, symbol: str) -> dict[str, Any]:
        data = await self._get(f"/stocks/{symbol.upper()}/company")
        company = data.get("company") or {}
        return company if isinstance(company, dict) else {}