base.py

"""Base class for token-related skills."""

import logging
from typing import Any, Dict, Optional

import aiohttp
from langchain_core.runnables import RunnableConfig

from abstracts.skill import SkillStoreABC
from skills.base import IntentKitSkill, SkillContext
from skills.token.constants import MORALIS_API_BASE_URL

logger = logging.getLogger(__name__)


class TokenBaseTool(IntentKitSkill):
    """Base class for all token-related skills.

    This base class provides common functionality for token API interactions,
    including making HTTP requests to the Moralis API.
    """

    def __init__(self, skill_store: SkillStoreABC = None):
        """Initialize the token tool with a skill store."""
        super().__init__(skill_store=skill_store)

    @property
    def category(self) -> str:
        return "token"

    def get_api_key(self, context: SkillContext) -> str:
        """Get API key from agent config or system config.

        Args:
            context: The skill context containing the agent config

        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":
            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.

        Raises:
            ValueError: If config is invalid or missing required fields
        """
        if not config:
            raise ValueError("No config provided to context_from_config")

        if "configurable" not in config:
            raise ValueError("'configurable' not in config")

        if "agent" not in config["configurable"]:
            raise ValueError("'agent' not in config['configurable']")

        agent = config["configurable"].get("agent")
        category_config = None

        if agent.skills:
            category_config = agent.skills.get(self.category)

        if not category_config:
            category_config = getattr(agent, self.category + "_config", {})

        if not category_config:
            category_config = {}

        from skills.base import SkillContext

        context = SkillContext(
            agent=agent,
            config=category_config,
            user_id=config["configurable"].get("user_id"),
            entrypoint=config["configurable"].get("entrypoint"),
        )

        return context

    def _prepare_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Convert boolean values to lowercase strings for API compatibility.

        Args:
            params: Dictionary with query parameters that may contain boolean values

        Returns:
            Dictionary with boolean values converted to lowercase strings
        """
        if not params:
            return params

        result = {}
        for key, value in params.items():
            if isinstance(value, bool):
                result[key] = str(value).lower()
            else:
                result[key] = value
        return result

    async def _make_request(
        self,
        method: str,
        endpoint: str,
        api_key: str,
        params: Dict[str, Any] = None,
        data: Dict[str, Any] = None,
    ) -> Dict[str, Any]:
        """Make a request to the Moralis API.

        Args:
            method: HTTP method (GET, POST, etc.)
            endpoint: API endpoint (without base URL)
            api_key: Moralis API key
            params: Query parameters
            data: Request body data for POST requests

        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:
            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

        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()