Feat/ Web3 and General tools

app/core/engine.py

@@ -57,6 +57,7 @@
 from app.core.graph import create_agent
 from app.core.prompt import agent_prompt
 from app.core.skill import skill_store
+from app.core.system import SystemStore
 from clients import TwitterClient
 from models.agent import Agent, AgentData
 from models.chat import AuthorType, ChatMessage, ChatMessageSkillCall
@@ -68,12 +69,14 @@
 from skills.common import get_common_skill
 from skills.elfa import get_elfa_skill
 from skills.enso import get_enso_skill
+from skills.general import get_general_skills
 from skills.goat import (
     create_smart_wallets_if_not_exist,
     get_goat_skill,
     init_smart_wallets,
 )
 from skills.twitter import get_twitter_skill
+from skills.w3 import get_web3_skill
 
 logger = logging.getLogger(__name__)
 
@@ -107,6 +110,10 @@ async def initialize_agent(aid, is_private=False):
         HTTPException: If agent not found (404) or database error (500)
     """
     """Initialize the agent with CDP Agentkit."""
+
+    # init global store
+    system_store = SystemStore()
+
     # init agent store
     agent_store = AgentStore(aid)
 
@@ -290,6 +297,25 @@ async def initialize_agent(aid, is_private=False):
                 except Exception as e:
                     logger.warning(e)
 
+    if (
+        hasattr(config, "chain_provider")
+        and agent.tx_skills
+        and len(agent.tx_skills) > 0
+    ):
+        for skill in agent.tx_skills:
+            try:
+                s = get_web3_skill(
+                    skill,
+                    config.chain_provider,
+                    system_store,
+                    skill_store,
+                    agent_store,
+                    aid,
+                )
+                tools.append(s)
+            except Exception as e:
+                logger.warning(e)
+
     # Enso skills
     if agent.enso_skills and len(agent.enso_skills) > 0 and agent.enso_config:
         for skill in agent.enso_skills:
@@ -373,6 +399,8 @@ async def initialize_agent(aid, is_private=False):
         for skill in agent.common_skills:
             tools.append(get_common_skill(skill))
 
+    tools.extend(get_general_skills(system_store, skill_store, agent_store, aid))
+
     # filter the duplicate tools
     tools = list({tool.name: tool for tool in tools}.values())
 

app/core/system.py

@@ -0,0 +1,15 @@
+from typing import Optional
+
+from models.w3 import W3Token
+from utils.chain import Network
+
+
+class SystemStore:
+    def __init__(self) -> None:
+        pass
+
+    async def get_token(self, symbol: str, network: Network) -> Optional[W3Token]:
+        return await W3Token.get(symbol, network)
+
+    async def get_all_wellknown_tokens(self) -> list[W3Token]:
+        return await W3Token.get_well_known()

docs/init_tokens.py

@@ -0,0 +1,104 @@
+import asyncio
+import json
+import time
+
+import requests
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.orm import sessionmaker
+from sqlmodel import select
+
+from models.w3 import W3Token
+
+# Database URL (replace with your actual database URL)
+DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/intentkit"
+
+# Create asynchronous engine
+engine = create_async_engine(DATABASE_URL)
+
+# Create session factory
+async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+
+async def get_session() -> AsyncSession:
+    async with async_session_maker() as session:
+        yield session
+
+
+base_url = "https://api.enso.finance/api/v1/tokens"
+headers = {
+    "accept": "application/json",
+    "Authorization": "Bearer 1e02632d-6feb-4a75-a157-documentation",
+}
+chain_ids = [
+    # 1, # Ethereum Mainnet
+    8453,
+    # 42161, # Arbitrum
+]
+
+
+async def add_to_db(item):
+    sym = item.get("symbol")
+    if sym and sym.strip() != "":
+        chid = str(item.get("chainId"))
+        async for db in get_session():
+            existing = (
+                await db.execute(
+                    select(W3Token).where(
+                        W3Token.symbol == sym,
+                        W3Token.chain_id == chid,
+                    )
+                )
+            ).first()
+            if not existing:
+                new_token = W3Token(
+                    symbol=sym,
+                    chain_id=chid,
+                    name=item.get("name"),
+                    decimals=item.get("decimals"),
+                    address=item.get("address"),
+                    primary_address=item.get("primaryAddress"),
+                    is_well_known=False,
+                    protocol_slug=item.get("protocolSlug"),
+                    token_type=item.get("type"),
+                )
+                db.add(new_token)
+
+            await db.commit()
+
+
+async def main():
+    for ch_id in chain_ids:
+        page = 1
+        while True:
+            url = f"{base_url}?&chainId={ch_id}&page={page}&includeMetadata=true"
+            try:
+                response = requests.get(url, headers=headers)
+                response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
+                data = response.json()
+                meta = data.get("meta", {})
+                tokens = data.get("data", [])  # access the items list
+                if (
+                    not tokens
+                ):  # if the items list is empty, then break out of the loop.
+                    break
+                print(f"processing chain {ch_id} page {page} of {meta["lastPage"]}")
+                for item in tokens:
+                    if item.get("underlyingTokens"):
+                        for t in item["underlyingTokens"]:
+                            await add_to_db(t)
+                    await add_to_db(item)
+                page += 1
+                if page > int(meta["lastPage"]):
+                    break
+                time.sleep(1)
+
+            except requests.exceptions.RequestException as e:
+                print(f"Error fetching page {page}: {e}")
+            except json.JSONDecodeError as e:
+                print(f"Error decoding JSON on page {page}: {e}")
+            except KeyError as e:
+                print(f"Error accessing 'items' in the JSON on page {page}: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

models/agent.py

@@ -113,6 +113,11 @@ class Agent(SQLModel, table=True):
     cdp_network_id: Optional[str] = Field(
         default="base-mainnet", description="Network identifier for CDP integration"
     )
+    tx_skills: Optional[List[str]] = Field(
+        default=None,
+        sa_column=Column(ARRAY(String)),
+        description="List of Transaction skills available to this agent",
+    )
     # if goat_enabled, will load goat skills
     crossmint_config: Optional[dict] = Field(
         default=None,

models/w3.py

@@ -0,0 +1,78 @@
+from typing import Optional
+
+from sqlmodel import Field, SQLModel, select
+
+from models.db import get_session
+from utils.chain import Network
+
+
+class W3Token(SQLModel, table=True):
+    """Model for storing token-specific data for web3 tools.
+
+    This model uses a composite primary key of (symbol, chain_id) to store
+    token data in a flexible way.
+
+    Attributes:
+        token_id: ID of the token
+        symbol: Token symbol
+    """
+
+    __tablename__ = "w3_token"
+
+    symbol: str = Field(primary_key=True)
+    chain_id: str = Field(primary_key=True)
+    is_well_known: bool = Field(nullable=False)
+    name: str = Field(nullable=False)
+    decimals: int = Field(nullable=False)
+    address: str = Field(nullable=False)
+    primary_address: str = Field(nullable=True)
+    token_type: str = Field(nullable=True)
+    protocol_slug: str = Field(nullable=True)
+
+    @classmethod
+    async def get(cls, symbol: str, network: Network) -> Optional["W3Token"]:
+        async with get_session() as db:
+            result = (
+                await db.exec(
+                    select(cls).where(
+                        cls.symbol == symbol,
+                        cls.chain_id == network.value.id,
+                    )
+                )
+            ).first()
+        return result
+
+    @classmethod
+    async def get_well_known(cls) -> list["W3Token"]:
+        async with get_session() as db:
+            result = (
+                await db.exec(
+                    select(cls).where(
+                        cls.is_well_known,
+                    )
+                )
+            ).all()
+        return result
+
+    async def save(self) -> None:
+        async with get_session() as db:
+            existing = (
+                await db.exec(
+                    select(self.__class__).where(
+                        self.__class__.symbol == self.symbol,
+                        self.__class__.chain_id == self.chain_id,
+                    )
+                )
+            ).first()
+            if existing:
+                existing.is_well_known = self.is_well_known
+                existing.name = self.name
+                existing.decimals = self.decimals
+                existing.address = self.address
+                existing.primary_address = self.primary_address
+                existing.token_type = self.token_type
+                existing.protocol_slug = self.protocol_slug
+                db.add(existing)
+            else:
+                db.add(self)
+            await db.commit()

pyproject.toml

@@ -68,6 +68,7 @@ langchain-xai = "^0.2.1"
 coinbase-agentkit = "0.1.4.dev202502250"
 coinbase-agentkit-langchain = "^0.1.0"
 jsonref = "^1.1.0"
+dateparser = "^1.2.1"
 
 [tool.poetry.group.dev] 
 optional = true

skills/enso/base.py

@@ -5,10 +5,10 @@
 
 from abstracts.agent import AgentStoreABC
 from abstracts.skill import IntentKitSkill, SkillStoreABC
-from utils.chain import ChainProvider, NetworkId
+from utils.chain import ChainProvider, Network
 
 base_url = "https://api.enso.finance"
-default_chain_id = int(NetworkId.BaseMainnet)
+default_chain_id = int(Network.BaseMainnet.value.id)
 
 
 class EnsoBaseTool(IntentKitSkill):

skills/general/__init__.py

@@ -0,0 +1,29 @@
+"""general skills."""
+
+from abstracts.skill import SkillStoreABC
+from app.core.system import SystemStore
+
+from .base import GeneralBaseTool
+from .timestamp import CurrentEpochTimestampTool, GetRelativeTimeParser
+
+
+def get_general_skills(
+    system_store: SystemStore,
+    skill_store: SkillStoreABC,
+    agent_store: SkillStoreABC,
+    agent_id: str,
+) -> list[GeneralBaseTool]:
+    return [
+        CurrentEpochTimestampTool(
+            agent_id=agent_id,
+            system_store=system_store,
+            skill_store=skill_store,
+            agent_store=agent_store,
+        ),
+        GetRelativeTimeParser(
+            agent_id=agent_id,
+            system_store=system_store,
+            skill_store=skill_store,
+            agent_store=agent_store,
+        ),
+    ]