TheGraph MCP Server

Instructions

Implementation Reference

• main.py:40-88 (handler)

  The handler function decorated with @mcp.tool(), implementing the core logic: performs GraphQL introspection query on TheGraph API for the given subgraphId, optionally converts JSON schema to GraphQL text using helper, handles errors.

  @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-38 (helper)

  Supporting utility function called by the handler when asText=True to convert the introspection JSON schema into human-readable GraphQL Schema Definition Language (SDL) text.

  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()