querySubgraph
Execute GraphQL queries against specific subgraphs on TheGraph MCP Server to retrieve indexed blockchain data in JSON format.
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 |
|---|---|---|---|
| query | Yes | ||
| subgraphId | Yes |
Implementation Reference
- main.py:89-110 (handler)The main handler function for the 'querySubgraph' tool, decorated with @mcp.tool() for registration. It checks for API key, constructs the API URL for the given subgraphId, posts the GraphQL query using httpx, and returns the JSON response or error.@mcp.tool() 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)}"