querySubgraph
Execute GraphQL queries against blockchain subgraphs to retrieve indexed data in JSON format for analysis and integration.
Instructions
Execute a GraphQL query against a specified subgraph.
Args: subgraphId (str): The ID of the subgraph to query. query (str): The GraphQL query string to execute.
Returns: str: Query result in JSON format, or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| subgraphId | Yes | ||
| query | Yes |
Implementation Reference
- main.py:90-110 (handler)The handler function for the 'querySubgraph' tool. It is decorated with @mcp.tool() for registration. It sends a POST request with the provided GraphQL query to the subgraph endpoint on TheGraph API and returns the JSON response or error.
async def querySubgraph(subgraphId: str, query: str) -> str: """Execute a GraphQL query against a specified subgraph. Args: subgraphId (str): The ID of the subgraph to query. query (str): The GraphQL query string to execute. Returns: str: Query result in JSON format, or an error message. """ if not API_KEY: return "API key is required. Set THEGRAPH_API_KEY in your .env file." async with httpx.AsyncClient() as client: url = f"{THEGRAPH_API_BASE_URL}{API_KEY}/subgraphs/id/{subgraphId}" try: response = await client.post(url, json={"query": query}, timeout=10) response.raise_for_status() return json.dumps(response.json()) except httpx.HTTPError as e: return f"Error executing query: {str(e)}"