diff --git a/skills/token/base.py b/skills/token/base.py index 64833fc03..e64924eb0 100644 --- a/skills/token/base.py +++ b/skills/token/base.py @@ -36,25 +36,38 @@ def get_api_key(self, context: SkillContext) -> str: Returns: The API key to use for API requests + + Raises: + ValueError: If no API key is available """ skill_config = context.config if skill_config.get("api_key_provider") == "agent_owner": - return skill_config.get("api_key") - return self.skill_store.get_system_config("moralis_api_key") + api_key = skill_config.get("api_key") + if not api_key: + raise ValueError( + "No agent-specific API key provided in the configuration." + ) + return api_key + + api_key = self.skill_store.get_system_config("moralis_api_key") + if not api_key: + raise ValueError("No Moralis API key provided in the configuration.") + return api_key def context_from_config(self, config: Optional[RunnableConfig] = None) -> Any: - """Extract context from the runnable config.""" + """Extract context from the runnable config. + + Raises: + ValueError: If config is invalid or missing required fields + """ if not config: - logger.error("No config provided to context_from_config") - return None + raise ValueError("No config provided to context_from_config") if "configurable" not in config: - logger.error("'configurable' not in config") - return None + raise ValueError("'configurable' not in config") if "agent" not in config["configurable"]: - logger.error("'agent' not in config['configurable']") - return None + raise ValueError("'agent' not in config['configurable']") agent = config["configurable"].get("agent") category_config = None @@ -118,37 +131,31 @@ async def _make_request( Returns: Response data as dictionary + + Raises: + ValueError: If API key is missing + aiohttp.ClientError: For HTTP client errors + Exception: For API errors or unexpected errors """ url = f"{MORALIS_API_BASE_URL}{endpoint}" if not api_key: - logger.error("API key is missing") - return {"error": "API key is missing"} + raise ValueError("API key is missing") headers = {"accept": "application/json", "X-API-Key": api_key} processed_params = self._prepare_params(params) if params else None - try: - async with aiohttp.ClientSession() as session: - async with session.request( - method=method, - url=url, - headers=headers, - params=processed_params, - json=data, - ) as response: - if response.status >= 400: - error_text = await response.text() - logger.error(f"API error {response.status}: {error_text}") - return { - "error": f"API error: {response.status}", - "details": error_text, - } - - return await response.json() - except aiohttp.ClientError as e: - logger.error(f"HTTP error making request: {str(e)}") - return {"error": f"HTTP error: {str(e)}"} - except Exception as e: - logger.error(f"Unexpected error making request: {str(e)}") - return {"error": f"Unexpected error: {str(e)}"} + async with aiohttp.ClientSession() as session: + async with session.request( + method=method, + url=url, + headers=headers, + params=processed_params, + json=data, + ) as response: + if response.status >= 400: + error_text = await response.text() + logger.error(f"API error {response.status}: {error_text}") + raise Exception(f"API error {response.status}: {error_text}") + + return await response.json() diff --git a/skills/token/erc20_transfers.py b/skills/token/erc20_transfers.py index 2edca298f..eb6b4c1eb 100644 --- a/skills/token/erc20_transfers.py +++ b/skills/token/erc20_transfers.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, List, Optional, Type +from typing import Any, Dict, List, Optional from langchain_core.runnables import RunnableConfig from pydantic import BaseModel, Field @@ -66,7 +66,7 @@ class ERC20Transfers(TokenBaseTool): "Get ERC20 token transactions for a wallet address, ordered by block number. " "Returns transaction details, token information, and wallet interactions." ) - args_schema: Type[BaseModel] = ERC20TransfersInput + args_schema: type[BaseModel] = ERC20TransfersInput async def _arun( self, @@ -102,19 +102,10 @@ async def _arun( Dict containing ERC20 transfer data """ context = self.context_from_config(config) - if context is None: - logger.error("Context is None, cannot retrieve API key") - return { - "error": "Cannot retrieve API key. Please check agent configuration." - } # Get the API key api_key = self.get_api_key(context) - if not api_key: - logger.error("No Moralis API key available") - return {"error": "No Moralis API key provided in the configuration."} - # Build query parameters params = {"chain": chain, "limit": limit, "order": order} @@ -133,13 +124,7 @@ async def _arun( params["cursor"] = cursor # Call Moralis API - try: - endpoint = f"/{address}/erc20/transfers" - return await self._make_request( - method="GET", endpoint=endpoint, api_key=api_key, params=params - ) - except Exception as e: - logger.error(f"Error fetching ERC20 transfers: {e}") - return { - "error": f"An error occurred while fetching ERC20 transfers: {str(e)}. Please try again later." - } + endpoint = f"/{address}/erc20/transfers" + return await self._make_request( + method="GET", endpoint=endpoint, api_key=api_key, params=params + ) diff --git a/skills/token/token_analytics.py b/skills/token/token_analytics.py index 4a06c0f22..9c9bda96e 100644 --- a/skills/token/token_analytics.py +++ b/skills/token/token_analytics.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, Type +from typing import Any, Dict from langchain_core.runnables import RunnableConfig from pydantic import BaseModel, Field @@ -32,7 +32,7 @@ class TokenAnalytics(TokenBaseTool): "Get analytics for a token by token address. " "Returns trading volumes, number of buyers/sellers, and liquidity information over various time periods." ) - args_schema: Type[BaseModel] = TokenAnalyticsInput + args_schema: type[BaseModel] = TokenAnalyticsInput async def _arun( self, @@ -52,30 +52,15 @@ async def _arun( Dict containing token analytics data """ context = self.context_from_config(config) - if context is None: - logger.error("Context is None, cannot retrieve API key") - return { - "error": "Cannot retrieve API key. Please check agent configuration." - } # Get the API key api_key = self.get_api_key(context) - if not api_key: - logger.error("No Moralis API key available") - return {"error": "No Moralis API key provided in the configuration."} - # Build query parameters params = {"chain": chain} # Call Moralis API - try: - endpoint = f"/tokens/{address}/analytics" - return await self._make_request( - method="GET", endpoint=endpoint, api_key=api_key, params=params - ) - except Exception as e: - logger.error(f"Error fetching token analytics: {e}") - return { - "error": f"An error occurred while fetching token analytics: {str(e)}. Please try again later." - } + endpoint = f"/tokens/{address}/analytics" + return await self._make_request( + method="GET", endpoint=endpoint, api_key=api_key, params=params + ) diff --git a/skills/token/token_price.py b/skills/token/token_price.py index bf657d8c4..b193fe464 100644 --- a/skills/token/token_price.py +++ b/skills/token/token_price.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, Optional, Type +from typing import Any, Dict, Optional from langchain_core.runnables import RunnableConfig from pydantic import BaseModel, Field @@ -54,7 +54,7 @@ class TokenPrice(TokenBaseTool): "Get the token price denominated in the blockchain's native token and USD for a given token contract address. " "Returns price, token information and exchange data." ) - args_schema: Type[BaseModel] = TokenPriceInput + args_schema: type[BaseModel] = TokenPriceInput async def _arun( self, @@ -86,19 +86,9 @@ async def _arun( # Extract context from config context = self.context_from_config(config) - if context is None: - logger.error("Context is None, cannot retrieve API key") - return { - "error": "Cannot retrieve API key. Please check agent configuration." - } - # Get the API key api_key = self.get_api_key(context) - if not api_key: - logger.error("No Moralis API key available") - return {"error": "No Moralis API key provided in the configuration."} - # Build query parameters params = {"chain": chain} @@ -115,18 +105,7 @@ async def _arun( params["min_pair_side_liquidity_usd"] = min_pair_side_liquidity_usd # Call Moralis API - try: - endpoint = f"/erc20/{address}/price" - response = await self._make_request( - method="GET", endpoint=endpoint, api_key=api_key, params=params - ) - - if "error" in response: - logger.error(f"API returned error: {response.get('error')}") - - return response - except Exception as e: - logger.error(f"Error fetching token price: {e}") - return { - "error": f"An error occurred while fetching token price: {str(e)}. Please try again later." - } + endpoint = f"/erc20/{address}/price" + return await self._make_request( + method="GET", endpoint=endpoint, api_key=api_key, params=params + ) diff --git a/skills/token/token_search.py b/skills/token/token_search.py index f3935bbcb..07450d25e 100644 --- a/skills/token/token_search.py +++ b/skills/token/token_search.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, List, Optional, Type +from typing import Any, Dict, List, Optional from langchain_core.runnables import RunnableConfig from pydantic import BaseModel, Field @@ -45,7 +45,7 @@ class TokenSearch(TokenBaseTool): "Returns token information including price, market cap, and security information. " "NOTE: This is a premium endpoint that requires a Moralis Business plan." ) - args_schema: Type[BaseModel] = TokenSearchInput + args_schema: type[BaseModel] = TokenSearchInput async def _arun( self, @@ -70,19 +70,10 @@ async def _arun( """ # Extract context from config context = self.context_from_config(config) - if context is None: - logger.error("Context is None, cannot retrieve API key") - return { - "error": "Cannot retrieve API key. Please check agent configuration." - } # Get the API key api_key = self.get_api_key(context) - if not api_key: - logger.error("No Moralis API key available") - return {"error": "No Moralis API key provided in the configuration."} - # Build query parameters params = {"query": query} @@ -95,27 +86,7 @@ async def _arun( params["isVerifiedContract"] = is_verified_contract # Call Moralis API - try: - endpoint = "/tokens/search" - result = await self._make_request( - method="GET", endpoint=endpoint, api_key=api_key, params=params - ) - - # Add premium notice if there's an error that might be related to plan limits - if "error" in result and "403" in str(result.get("error", "")): - logger.error("Received 403 error - likely a plan limitation") - result["notice"] = ( - "This API requires a Moralis Business plan or Enterprise plan. " - "Please ensure your API key is associated with the appropriate plan." - ) - - return result - except Exception as e: - logger.error(f"Error searching for tokens: {e}") - return { - "error": f"An error occurred while searching for tokens: {str(e)}. Please try again later.", - "notice": ( - "This API requires a Moralis Business plan or Enterprise plan. " - "Please ensure your API key is associated with the appropriate plan." - ), - } + endpoint = "/tokens/search" + return await self._make_request( + method="GET", endpoint=endpoint, api_key=api_key, params=params + )