ens-mcp

Instructions

Implementation Reference

• main.py:229-267 (handler)

  The primary handler function for the 'get_domain_events' MCP tool. It is decorated with @mcp.tool() for registration, calls the helper query_domain_events, processes the events by type, and returns a formatted string summary.
  @mcp.tool() async def get_domain_events(domain: str) -> str: """Retrieve events associated with an ENS domain.""" events = await query_domain_events(domain) if not events: return f"No events found for ENS domain: {domain}" event_summaries = [] for event in events: event_type = event["__typename"] summary = f"Event: {event_type}\n" summary += f"Block Number: {event['blockNumber']}\n" summary += f"Transaction ID: {event['transactionID']}\n" if event_type == "Transfer": summary += f"New Owner: {event['owner']['id']}" elif event_type == "NewOwner": summary += f"New Owner: {event['owner']['id']}\nParent Domain: {event['parentDomain']['name']}" elif event_type == "NewResolver": addr = event['resolver']['addr']['id'] if event['resolver']['addr'] else "None" summary += f"Resolver Address: {event['resolver']['address']}\nResolver Addr: {addr}" elif event_type == "NewTTL": summary += f"TTL: {event['ttl']} seconds" elif event_type == "WrappedTransfer": summary += f"New Wrapped Owner: {event['owner']['id']}" elif event_type == "NameWrapped": expiry = datetime.datetime.fromtimestamp(int(event['expiryDate'])).strftime("%Y-%m-%d %H:%M:%S") summary += f"Wrapped Owner: {event['owner']['id']}\nName: {event['name']}\nFuses: {event['fuses']}\nExpiry: {expiry}" elif event_type == "NameUnwrapped": summary += f"Owner: {event['owner']['id']}" elif event_type == "FusesSet": summary += f"Fuses: {event['fuses']}" elif event_type == "ExpiryExtended": expiry = datetime.datetime.fromtimestamp(int(event['expiryDate'])).strftime("%Y-%m-%d %H:%M:%S") summary += f"New Expiry: {expiry}" event_summaries.append(summary) return "\n\n".join(event_summaries)
• main.py:84-148 (helper)

  Supporting helper function that executes the GraphQL query to retrieve raw domain events from the ENS subgraph for the specified domain name.
  async def query_domain_events(name: str) -> List[Dict[str, Any]]: """Query the ENS Subgraph for domain events.""" query = gql(""" query GetDomainEvents($name: String!) { domains(where: { name: $name }) { events { id __typename blockNumber transactionID ... on Transfer { owner { id } } ... on NewOwner { owner { id } parentDomain { name } } ... on NewResolver { resolver { address addr { id } } } ... on NewTTL { ttl } ... on WrappedTransfer { owner { id } } ... on NameWrapped { owner { id } name fuses expiryDate } ... on NameUnwrapped { owner { id } } ... on FusesSet { fuses } ... on ExpiryExtended { expiryDate } } } } """) result = await graphql_client.execute_async(query, variable_values={"name": name}) return result["domains"][0]["events"] if result["domains"] and result["domains"][0]["events"] else []
• main.py:86-145 (schema)

  Embedded GraphQL schema/query definition (GetDomainEvents) that defines the structure and fields for fetching domain events data from the ENS subgraph.
  query = gql(""" query GetDomainEvents($name: String!) { domains(where: { name: $name }) { events { id __typename blockNumber transactionID ... on Transfer { owner { id } } ... on NewOwner { owner { id } parentDomain { name } } ... on NewResolver { resolver { address addr { id } } } ... on NewTTL { ttl } ... on WrappedTransfer { owner { id } } ... on NameWrapped { owner { id } name fuses expiryDate } ... on NameUnwrapped { owner { id } } ... on FusesSet { fuses } ... on ExpiryExtended { expiryDate } } } } """)