get_domain_events

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
            }
          }
        }
      }
  """)