Merge pull request #902 from crestalnetwork/codex/add-function-to-convert-ens-domain-to-wallet-address

feat: add ENS resolution utility

Author: hyacinthus

intentkit/models/agent.py

@@ -38,11 +38,18 @@
 from intentkit.models.db import get_session
 from intentkit.models.llm import LLMModelInfo, LLMProvider
 from intentkit.models.skill import Skill
+from intentkit.utils.ens import resolve_ens_to_address
 from intentkit.utils.error import IntentKitAPIError
 
 logger = logging.getLogger(__name__)
 
 
+ENS_NAME_PATTERN = re.compile(
+    r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+(?:eth|base\.eth)$",
+    re.IGNORECASE,
+)
+
+
 class AgentAutonomous(BaseModel):
     """Autonomous agent configuration."""
 
@@ -1337,16 +1344,20 @@ async def get_by_id_or_slug(cls, agent_id: str) -> "Agent" | None:
         Returns:
             Agent if found, None otherwise
         """
+        query_id = agent_id
+        if ENS_NAME_PATTERN.fullmatch(agent_id):
+            query_id = await resolve_ens_to_address(agent_id)
+
         async with get_session() as db:
             agent = None
 
             # Try to get by ID if length <= 20
-            if len(agent_id) <= 20:
-                agent = await Agent.get(agent_id)
+            if len(query_id) <= 20 or query_id.startswith("0x"):
+                agent = await Agent.get(query_id)
 
             # If not found, try to get by slug
             if agent is None:
-                slug_stmt = select(AgentTable).where(AgentTable.slug == agent_id)
+                slug_stmt = select(AgentTable).where(AgentTable.slug == query_id)
                 agent_row = await db.scalar(slug_stmt)
                 if agent_row is not None:
                     agent = Agent.model_validate(agent_row)

intentkit/utils/__init__.py

@@ -1 +1,5 @@
+"""Utility exports for IntentKit."""
 
+from intentkit.utils.ens import resolve_ens_to_address
+
+__all__ = ["resolve_ens_to_address"]

intentkit/utils/ens.py

@@ -0,0 +1,125 @@
+"""Utilities for resolving ENS domains to wallet addresses."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from ens import ENS
+from web3 import Web3
+from web3.middleware import geth_poa_middleware
+
+from intentkit.config.config import config
+from intentkit.models.redis import get_redis
+from intentkit.utils.error import IntentKitAPIError
+
+logger = logging.getLogger(__name__)
+
+_CACHE_PREFIX = "intentkit:ens:"
+_CACHE_TTL_SECONDS = 4 * 60 * 60
+
+_NETWORKS_BY_SUFFIX: dict[str, tuple[str, ...]] = {
+    ".base.eth": ("base-mainnet", "ethereum-mainnet"),
+    ".eth": ("ethereum-mainnet",),
+}
+
+
+async def resolve_ens_to_address(name: str) -> str:
+    """Resolve an ENS domain to a checksum wallet address.
+
+    Args:
+        name: ENS name to resolve.
+
+    Returns:
+        The checksum wallet address associated with the ENS name.
+
+    Raises:
+        IntentKitAPIError: If the ENS name cannot be resolved to a wallet address.
+    """
+
+    normalized = name.strip().lower()
+    if not normalized:
+        raise IntentKitAPIError(404, "ENSNameNotFound", "ENS name is empty.")
+
+    redis_client = None
+    cache_key = f"{_CACHE_PREFIX}{normalized}"
+    try:
+        redis_client = get_redis()
+    except Exception:  # Redis is optional; ignore if unavailable
+        redis_client = None
+
+    if redis_client is not None:
+        cached_address = await redis_client.get(cache_key)
+        if cached_address:
+            return cached_address
+
+    networks = _networks_for_name(normalized)
+    if not networks:
+        raise IntentKitAPIError(
+            404,
+            "ENSNameNotFound",
+            "Unsupported ENS name suffix.",
+        )
+
+    for network in networks:
+        address = await _resolve_on_network(normalized, network)
+        if address:
+            if redis_client is not None:
+                try:
+                    await redis_client.set(cache_key, address, ex=_CACHE_TTL_SECONDS)
+                except Exception as exc:  # pragma: no cover - optional cache
+                    logger.debug("Failed to cache ENS resolution: %s", exc)
+            return address
+
+    raise IntentKitAPIError(
+        404,
+        "ENSNameNotFound",
+        f"ENS name {name} could not be resolved.",
+    )
+
+
+def _networks_for_name(name: str) -> tuple[str, ...]:
+    for suffix, networks in _NETWORKS_BY_SUFFIX.items():
+        if name.endswith(suffix):
+            return networks
+    return tuple()
+
+
+async def _resolve_on_network(name: str, network: str) -> str | None:
+    chain_provider = getattr(config, "chain_provider", None)
+    if chain_provider is None:
+        logger.debug("No chain provider configured; cannot resolve ENS name.")
+        return None
+
+    try:
+        chain_config = chain_provider.get_chain_config(network)
+    except Exception as exc:  # pragma: no cover - dependent on external config
+        logger.debug("Chain config for %s unavailable: %s", network, exc)
+        return None
+
+    rpc_url = chain_config.ens_url or chain_config.rpc_url
+    if not rpc_url:
+        logger.debug("No RPC/ENS URL configured for %s", network)
+        return None
+
+    def _resolve() -> str | None:
+        web3_client = Web3(Web3.HTTPProvider(rpc_url))
+        if network.startswith("base"):
+            web3_client.middleware_onion.inject(geth_poa_middleware, layer=0)
+
+        ens_client = ENS.from_web3(web3_client)
+        try:
+            resolved = ens_client.address(name)
+        except Exception as exc:  # pragma: no cover - dependent on external provider
+            logger.debug("Error resolving %s on %s: %s", name, network, exc)
+            return None
+
+        if not resolved:
+            return None
+
+        try:
+            return Web3.to_checksum_address(resolved)
+        except ValueError:
+            return None
+
+    return await asyncio.to_thread(_resolve)