feat: add fetch_json util

HenryQW · HenryQW · commit edf56166741f · 2025-02-27T16:13:28.000+08:00
diff --git a/openagent/core/utils/fetch_json.py b/openagent/core/utils/fetch_json.py
@@ -0,0 +1,42 @@
+from typing import Dict, Optional, Any
+
+import httpx
+from loguru import logger
+
+
+async def fetch_json(
+    url: str,
+    method: str = "GET",
+    params: Optional[Dict[str, Any]] = None,
+    headers: Optional[Dict[str, str]] = None,
+    timeout: float = 30.0,
+    **kwargs,
+) -> Dict:
+    """
+    Fetches JSON data from a given URL using an HTTP request.
+
+    :param url: The API endpoint.
+    :param method: HTTP method ('GET', 'POST', etc.), defaults to 'GET'.
+    :param params: Optional query parameters.
+    :param headers: Optional request headers.
+    :param timeout: Request timeout in seconds (default 30.0).
+    :param kwargs: Additional arguments for request customization.
+    :return: JSON response as a dictionary.
+    :raises httpx.HTTPStatusError: If the response contains an error status.
+    :raises httpx.RequestError: If a network error occurs.
+    """
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        try:
+            response = await client.request(
+                method=method, url=url, params=params, headers=headers, **kwargs
+            )
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            logger.error(
+                f"HTTP error {e.response.status_code} for URL {url}: {e.response.text}"
+            )
+            raise
+        except httpx.RequestError as e:
+            logger.error(f"Request error for URL {url}: {e}")
+            raise
diff --git a/openagent/tools/arbitrum/compound_market_analysis.py b/openagent/tools/arbitrum/compound_market_analysis.py
@@ -10,6 +10,7 @@
 from pydantic import BaseModel, Field
 from openagent.agent.config import ModelConfig
 from openagent.core.tool import Tool
+from openagent.core.utils.fetch_json import fetch_json
 
 
 @dataclass
@@ -112,69 +113,51 @@ async def __call__(self) -> str:
 
     @staticmethod
     async def _fetch_compound_arbitrum_market_data() -> list[CompoundMarketData]:
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            response = await client.get(
-                "https://v3-api.compound.finance/market/all-networks/all-contracts/summary"
-            )
-
-            if response.status_code != 200:
-                raise Exception(
-                    f"Failed to fetch Compound market data: {response.text}, {response.status_code}"
-                )
-
-            results = response.json()
+        results = await fetch_json(
+            url="https://v3-api.compound.finance/market/all-networks/all-contracts/summary"
+        )
 
-            # Filter for Arbitrum markets (chain_id 42161)
-            arbitrum_markets = [
-                market for market in results if market["chain_id"] == 42161
-            ]
+        # Filter for Arbitrum markets (chain_id 42161)
+        arbitrum_markets = [market for market in results if market["chain_id"] == 42161]
 
-            market_data = []
+        market_data = []
 
-            for market in arbitrum_markets:
-                # Fetch historical data for each address
-                historical_response = await client.get(
-                    f"https://v3-api.compound.finance/market/arbitrum-mainnet/{market['comet']['address']}/historical/summary"
-                )
-
-                if historical_response.status_code != 200:
-                    logger.warning(
-                        f"Failed to fetch historical data for {market['comet']['address']}: {historical_response.text}, {historical_response.status_code}"
-                    )
-                    continue
+        for market in arbitrum_markets:
+            # Fetch historical data for each address
+            historical_data = await fetch_json(
+                f"https://v3-api.compound.finance/market/arbitrum-mainnet/{market['comet']['address']}/historical/summary"
+            )
 
-                historical_data = historical_response.json()
+            # Sort historical data by timestamp in descending order (newest first)
+            sorted_data = sorted(
+                historical_data, key=lambda x: x["timestamp"], reverse=True
+            )
 
-                # Sort historical data by timestamp in descending order (newest first)
-                sorted_data = sorted(
-                    historical_data, key=lambda x: x["timestamp"], reverse=True
+            if len(sorted_data) < 2:
+                logger.warning(
+                    f"Insufficient historical data for {market['comet']['address']}"
                 )
-
-                if len(sorted_data) < 2:
-                    logger.warning(
-                        f"Insufficient historical data for {market['comet']['address']}"
-                    )
-                    continue
-
-                # Convert string APRs to float
-                current_borrow_apr = float(sorted_data[0]["borrow_apr"])
-                current_supply_apr = float(sorted_data[0]["supply_apr"])
-                yesterday_borrow_apr = float(sorted_data[1]["borrow_apr"])
-                yesterday_supply_apr = float(sorted_data[1]["supply_apr"])
-
-                # Calculate 24h changes
-                borrow_apr_change_24h = current_borrow_apr - yesterday_borrow_apr
-                supply_apr_change_24h = current_supply_apr - yesterday_supply_apr
-
-                market_data.append(
-                    CompoundMarketData(
-                        address=market["comet"]["address"],
-                        collateralAssets=market["collateral_asset_symbols"],
-                        borrowAPR=current_borrow_apr,
-                        supplyAPR=current_supply_apr,
-                        borrowAPRChange24h=borrow_apr_change_24h,
-                        supplyAPRChange24h=supply_apr_change_24h,
-                    )
+                continue
+
+            # Convert string APRs to float
+            current_borrow_apr = float(sorted_data[0]["borrow_apr"])
+            current_supply_apr = float(sorted_data[0]["supply_apr"])
+            yesterday_borrow_apr = float(sorted_data[1]["borrow_apr"])
+            yesterday_supply_apr = float(sorted_data[1]["supply_apr"])
+
+            # Calculate 24h changes
+            borrow_apr_change_24h = current_borrow_apr - yesterday_borrow_apr
+            supply_apr_change_24h = current_supply_apr - yesterday_supply_apr
+
+            market_data.append(
+                CompoundMarketData(
+                    address=market["comet"]["address"],
+                    collateralAssets=market["collateral_asset_symbols"],
+                    borrowAPR=current_borrow_apr,
+                    supplyAPR=current_supply_apr,
+                    borrowAPRChange24h=borrow_apr_change_24h,
+                    supplyAPRChange24h=supply_apr_change_24h,
                 )
+            )
 
-            return market_data
+        return market_data
diff --git a/openagent/tools/pendle/market_analysis.py b/openagent/tools/pendle/market_analysis.py
@@ -19,6 +19,7 @@
 from openagent.agent.config import ModelConfig
 from openagent.core.database import sqlite
 from openagent.core.tool import Tool
+from openagent.core.utils.fetch_json import fetch_json
 
 Base = declarative_base()
 
@@ -234,25 +235,18 @@ async def __call__(self) -> str:
             return error_msg
 
     async def _fetch_pendle_market_data(self) -> PendleMarket:
-        async with httpx.AsyncClient() as client:
-            response = await client.get(
-                "https://api-v2.pendle.finance/bff/v3/markets/all?isActive=true"
-            )
-            if response.status_code != 200:
-                raise Exception(
-                    f"API request failed with status {response.status_code}"
-                )
-
-            # Get Pendle market data from API
-            results = response.json()
+        # Get Pendle market data from API
+        result = await fetch_json(
+            url="https://api-v2.pendle.finance/bff/v3/markets/all?isActive=true"
+        )
 
-            # Process the data
-            snapshot = self._process_market_data(results)
+        # Process the data
+        snapshot = self._process_market_data(result)
 
-            return PendleMarket(
-                data=json.dumps(asdict(snapshot)),
-                created_at=datetime.now(UTC),
-            )
+        return PendleMarket(
+            data=json.dumps(asdict(snapshot)),
+            created_at=datetime.now(UTC),
+        )
 
     @staticmethod
     def _process_market_data(results: dict) -> PendleMarketSnapshot:
diff --git a/openagent/tools/pendle/voter_apy_analysis.py b/openagent/tools/pendle/voter_apy_analysis.py
@@ -16,6 +16,7 @@
 from openagent.agent.config import ModelConfig
 from openagent.core.database import sqlite
 from openagent.core.tool import Tool
+from openagent.core.utils.fetch_json import fetch_json
 from openagent.core.utils.json_equal import json_equal
 
 Base = declarative_base()
@@ -149,28 +150,23 @@ async def __call__(self) -> str:
             return error_msg
 
     async def _fetch_pendle_voter_apy(self) -> PendleVoterApy:
-        async with httpx.AsyncClient() as client:
-            response = await client.get(
-                "https://api-v2.pendle.finance/bff/v1/ve-pendle/pool-voter-apy"
-            )
-            if response.status_code != 200:
-                raise Exception(
-                    f"API request failed with status {response.status_code}"
-                )
+        # Get Pendle voter data from API
+        result = await fetch_json(
+            url="https://api-v2.pendle.finance/bff/v1/ve-pendle/pool-voter-apy"
+        )
 
-            result = response.json()
-            if not result["results"]:
-                raise Exception("API response is empty")
+        if not result["results"]:
+            raise Exception("API response is empty")
 
-            data = self._filter_pendle_voter_apy(result)
+        data = self._filter_pendle_voter_apy(result)
 
-            # Create new snapshot
-            snapshot = PendleVoterApy(
-                data=str(data),
-                created_at=datetime.now(UTC),
-            )
+        # Create new snapshot
+        snapshot = PendleVoterApy(
+            data=str(data),
+            created_at=datetime.now(UTC),
+        )
 
-            return snapshot
+        return snapshot
 
     @staticmethod
     def _filter_pendle_voter_apy(apy_data: dict) -> dict:
@@ -186,9 +182,7 @@ def extract_pool_info(item: dict) -> dict:
             change_direction = (
                 "increased"
                 if last_epoch_change > 0
-                else "decreased"
-                if last_epoch_change < 0
-                else "unchanged"
+                else "decreased" if last_epoch_change < 0 else "unchanged"
             )
 
             # Convert to percentage strings, use absolute value for lastEpochChange
