getSubgraphSchema
Retrieve the schema of a specific subgraph in JSON or GraphQL text format using GraphQL introspection. Ideal for analyzing or integrating subgraph data structures with applications.
Instructions
Fetch the schema of a specified subgraph using GraphQL introspection.
Args: subgraphId (str): The ID of the subgraph to query. asText (bool): If True, return schema as GraphQL text; otherwise, return JSON.
Returns: str: Schema in JSON or GraphQL text format, or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| asText | No | ||
| subgraphId | Yes |
Implementation Reference
- main.py:40-88 (handler)The handler function for the 'getSubgraphSchema' tool. It performs GraphQL introspection on the specified subgraph using TheGraph API, processes the schema data, and returns it as JSON or GraphQL text format based on the 'asText' parameter.@mcp.tool() async def getSubgraphSchema(subgraphId: str, asText: bool = False) -> str: """Fetch the schema of a specified subgraph using GraphQL introspection. Args: subgraphId (str): The ID of the subgraph to query. asText (bool): If True, return schema as GraphQL text; otherwise, return JSON. Returns: str: Schema in JSON or GraphQL text 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}" introspection_query = """ query IntrospectionQuery { __schema { types { name kind fields { name type { name kind ofType { name kind } } } } } } """ try: response = await client.post(url, json={"query": introspection_query}, timeout=10) response.raise_for_status() schema_data = response.json() if "data" in schema_data and "__schema" in schema_data["data"]: schema = schema_data["data"]["__schema"] return json_to_graphql_schema(schema) if asText else json.dumps(schema) else: return f"Failed to fetch schema for {subgraphId}" except httpx.HTTPError as e: return f"Error fetching schema: {str(e)}"
- main.py:19-39 (helper)Helper function that converts the JSON introspection schema into a GraphQL schema text format, used when asText=True in the handler.def json_to_graphql_schema(schema_json): """Convert JSON schema from introspection to GraphQL text format.""" types = schema_json["types"] schema_text = "" for t in types: if t["kind"] == "OBJECT" and not t["name"].startswith("__"): schema_text += f"type {t['name']} {{\n" if t["fields"]: for f in t["fields"]: field_type = f["type"] type_name = field_type["name"] if field_type["kind"] == "NON_NULL": type_name = f"{field_type['ofType']['name']}!" elif field_type["kind"] == "LIST": type_name = f"[{field_type['ofType']['name']}]" schema_text += f" {f['name']}: {type_name}\n" schema_text += "}\n\n" return schema_text.strip()
- main.py:40-40 (registration)The @mcp.tool() decorator registers the getSubgraphSchema function as an MCP tool.@mcp.tool()
- main.py:41-50 (schema)Type annotations and docstring define the input schema (subgraphId: str, asText: bool=False) and output (str). FastMCP likely uses these for tool schema.async def getSubgraphSchema(subgraphId: str, asText: bool = False) -> str: """Fetch the schema of a specified subgraph using GraphQL introspection. Args: subgraphId (str): The ID of the subgraph to query. asText (bool): If True, return schema as GraphQL text; otherwise, return JSON. Returns: str: Schema in JSON or GraphQL text format, or an error message. """