main.py•9.6 kB
from mcp.server.fastmcp import FastMCP
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from typing import Dict, Any, Optional, List
import datetime
import os
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize MCP server
mcp = FastMCP("ens-mcp")
# GraphQL client setup
GRAPH_API_URL = "https://gateway.thegraph.com/api/subgraphs/id/5XqPmWe6gjyrJtFn9cLy237i4cWw2j9HcUJEXsP5qGtH"
API_KEY = os.getenv("THEGRAPH_API_KEY")
if not API_KEY:
raise ValueError("THEGRAPH_API_KEY not found in .env file")
transport = AIOHTTPTransport(
url=GRAPH_API_URL,
headers={"Authorization": f"Bearer {API_KEY}"}
)
graphql_client = Client(transport=transport, fetch_schema_from_transport=True)
async def query_ens_domain(name: str) -> Optional[Dict[str, Any]]:
"""Query the ENS Subgraph for domain details."""
query = gql("""
query GetDomain($name: String!) {
domains(where: { name: $name }) {
id
name
labelName
labelhash
subdomainCount
resolvedAddress {
id
}
resolver {
address
addr {
id
}
contentHash
texts
}
ttl
isMigrated
createdAt
owner {
id
}
registrant {
id
}
wrappedOwner {
id
}
expiryDate
registration {
registrationDate
expiryDate
cost
registrant {
id
}
labelName
}
wrappedDomain {
expiryDate
fuses
owner {
id
}
name
}
}
}
""")
result = await graphql_client.execute_async(query, variable_values={"name": name})
return result["domains"][0] if result["domains"] else None
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 []
@mcp.tool()
async def resolve_ens_name(domain: str) -> str:
"""Resolve an ENS name to its Ethereum address."""
domain_data = await query_ens_domain(domain)
if not domain_data:
return f"No data found for ENS domain: {domain}"
# Prefer resolvedAddress, fallback to resolver.addr
address = (domain_data["resolvedAddress"]["id"] if domain_data["resolvedAddress"]
else domain_data["resolver"]["addr"]["id"] if domain_data["resolver"] and domain_data["resolver"]["addr"]
else "None")
return address
@mcp.tool()
async def get_domain_details(domain: str) -> str:
"""Fetch detailed information for an ENS domain, including its address."""
domain_data = await query_ens_domain(domain)
if not domain_data:
return f"No data found for ENS domain: {domain}"
# Get address
address = (domain_data["resolvedAddress"]["id"] if domain_data["resolvedAddress"]
else domain_data["resolver"]["addr"]["id"] if domain_data["resolver"] and domain_data["resolver"]["addr"]
else "None")
# Format dates
expiry = (datetime.datetime.fromtimestamp(int(domain_data["expiryDate"]))
.strftime("%Y-%m-%d %H:%M:%S") if domain_data["expiryDate"]
else "None")
created = (datetime.datetime.fromtimestamp(int(domain_data["createdAt"]))
.strftime("%Y-%m-%d %H:%M:%S") if domain_data["createdAt"]
else "None")
# Registration details
registration_info = (
f"Registration Date: {datetime.datetime.fromtimestamp(int(domain_data['registration']['registrationDate'])).strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Registration Expiry: {datetime.datetime.fromtimestamp(int(domain_data['registration']['expiryDate'])).strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Registration Cost: {domain_data['registration']['cost'] or 'Unknown'} Wei\n"
f"Registrant: {domain_data['registration']['registrant']['id']}"
if domain_data["registration"]
else "No Registration"
)
# Wrapped domain details
wrapped_info = (
f"Wrapped Name: {domain_data['wrappedDomain']['name']}\n"
f"Wrapped Owner: {domain_data['wrappedDomain']['owner']['id']}\n"
f"Wrapped Expiry: {datetime.datetime.fromtimestamp(int(domain_data['wrappedDomain']['expiryDate'])).strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Fuses: {domain_data['wrappedDomain']['fuses']}"
if domain_data["wrappedDomain"]
else "Not Wrapped"
)
# Resolver details
resolver_info = (
f"Resolver Address: {domain_data['resolver']['address']}\n"
f"Content Hash: {domain_data['resolver']['contentHash'] or 'None'}\n"
f"Text Records: {', '.join(domain_data['resolver']['texts']) if domain_data['resolver']['texts'] else 'None'}"
if domain_data["resolver"]
else "No Resolver"
)
return (
f"ENS Domain: {domain_data['name']}\n"
f"Address: {address}\n"
f"Label Name: {domain_data['labelName'] or 'None'}\n"
f"Label Hash: {domain_data['labelhash'] or 'None'}\n"
f"Subdomain Count: {domain_data['subdomainCount']}\n"
f"Owner: {domain_data['owner']['id']}\n"
f"Registrant: {domain_data['registrant']['id'] if domain_data['registrant'] else 'None'}\n"
f"Wrapped Owner: {domain_data['wrappedOwner']['id'] if domain_data['wrappedOwner'] else 'None'}\n"
f"Expiry Date: {expiry}\n"
f"TTL: {domain_data['ttl'] or 'None'} seconds\n"
f"Is Migrated: {domain_data['isMigrated']}\n"
f"Created At: {created}\n"
f"Registration: {registration_info}\n"
f"Wrapped Domain: {wrapped_info}\n"
f"Resolver: {resolver_info}"
)
@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)
# Run the server
if __name__ == "__main__":
mcp.run()