Subgraph Registry MCP

Instructions

Implementation Reference

• src/index.js:83-170 (handler)

  JavaScript implementation of search_subgraphs tool. Queries SQLite database to filter and search subgraphs by domain, network, protocol type, entity, reliability score, and keywords. Implements deduplication by IPFS hash and returns ranked results with query URLs.

  function searchSubgraphs({
    query = "",
    domain = "",
    network = "",
    protocol_type = "",
    entity = "",
    min_reliability = 0,
    limit = 20,
  } = {}) {
    const conditions = [];
    const params = [];
  
    if (domain) {
      conditions.push("domain = ?");
      params.push(domain);
    }
    if (network) {
      conditions.push("network = ?");
      params.push(network);
    }
    if (protocol_type) {
      conditions.push("protocol_type = ?");
      params.push(protocol_type);
    }
    if (entity) {
      conditions.push('canonical_entities LIKE ?');
      params.push(`%"${entity}"%`);
    }
    if (min_reliability > 0) {
      conditions.push("reliability_score >= ?");
      params.push(min_reliability);
    }
    if (query) {
      const words = query.trim().split(/\s+/).filter((w) => w.length > 2).slice(0, 5);
      if (words.length) {
        const wordConds = words.map(() => "(display_name LIKE ? OR description LIKE ? OR auto_description LIKE ?)");
        words.forEach((w) => params.push(`%${w}%`, `%${w}%`, `%${w}%`));
        conditions.push(`(${wordConds.join(" OR ")})`);
      } else {
        conditions.push("(display_name LIKE ? OR description LIKE ? OR auto_description LIKE ?)");
        params.push(`%${query}%`, `%${query}%`, `%${query}%`);
      }
    }
  
    const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
    // Over-fetch to allow dedup by IPFS hash (same deployment, different subgraph IDs)
    const fetchLimit = limit * 3;
    const sql = `
      SELECT id, display_name, description, auto_description, domain, protocol_type, network,
             reliability_score, ipfs_hash, entity_count, canonical_entities,
             powered_by_substreams
      FROM subgraphs
      ${where}
      ORDER BY reliability_score DESC
      LIMIT ?
    `;
    params.push(fetchLimit);
  
    const rows = getDb().prepare(sql).all(...params);
    // Dedup by IPFS hash — keep highest reliability per deployment
    const seenIpfs = new Set();
    const results = [];
    for (const r of rows) {
      if (r.ipfs_hash && seenIpfs.has(r.ipfs_hash)) continue;
      if (r.ipfs_hash) seenIpfs.add(r.ipfs_hash);
      results.push({
        id: r.id,
        display_name: r.display_name,
        description: (r.description || r.auto_description || "").slice(0, 300),
        domain: r.domain,
        protocol_type: r.protocol_type,
        network: r.network,
        reliability_score: r.reliability_score,
        ipfs_hash: r.ipfs_hash,
        entity_count: r.entity_count,
        canonical_entities: JSON.parse(r.canonical_entities),
        powered_by_substreams: Boolean(r.powered_by_substreams),
        query_url: `https://gateway.thegraph.com/api/[api-key]/subgraphs/id/${r.id}`,
      });
      if (results.length >= limit) break;
    }
  
    return {
      total: results.length,
      subgraphs: results,
      query_instructions: "To query a subgraph: POST a GraphQL query to the query_url (replace [api-key] with your Graph API key from https://thegraph.com/studio/apikeys/). First fetch the schema with get_subgraph_detail to see available entities and fields.",
    };
  }

• src/index.js:318-333 (schema)

  MCP tool schema definition for search_subgraphs. Defines input parameters (query, domain, network, protocol_type, entity, min_reliability, limit) with types and descriptions for filtering the subgraph registry.

    name: "search_subgraphs",
    description:
      "Search and filter the classified subgraph registry (15,500+ subgraphs). Filter by domain (defi, nfts, dao, gaming, identity, infrastructure, social, analytics), network (mainnet, arbitrum-one, base, matic, bsc, optimism, avalanche), protocol_type (dex, lending, bridge, staking, options, perpetuals, nft-marketplace, yield-aggregator, governance, name-service), canonical entity type (liquidity_pool, trade, token, position, vault, loan, collateral, liquidation, nft_collection, nft_item, nft_sale, proposal, delegate, domain_name, account, transaction, daily_snapshot, hourly_snapshot), or free-text keyword. Returns subgraphs ranked by reliability score with query URLs. To query data: POST GraphQL to https://gateway.thegraph.com/api/[api-key]/subgraphs/id/[subgraph-id] (get API key from https://thegraph.com/studio/apikeys/).",
    inputSchema: {
      type: "object",
      properties: {
        query: { type: "string", description: "Free-text search across names and descriptions" },
        domain: { type: "string", description: "Filter by domain: defi, nfts, dao, gaming, identity, infrastructure, social, analytics" },
        network: { type: "string", description: "Filter by chain: mainnet, arbitrum-one, base, matic, bsc, optimism, avalanche, etc." },
        protocol_type: { type: "string", description: "Filter by protocol type: dex, lending, bridge, staking, options, perpetuals, etc." },
        entity: { type: "string", description: "Filter by canonical entity: liquidity_pool, trade, token, position, vault, loan, etc." },
        min_reliability: { type: "number", description: "Minimum reliability score (0-1). Higher = more signal/stake/fees." },
        limit: { type: "integer", description: "Max results to return (default: 20)", default: 20 },
      },
    },
  },

• src/index.js:370-375 (registration)

  Registration of search_subgraphs handler in the HANDLERS object, mapping tool name to the searchSubgraphs function implementation.

  const HANDLERS = {
    search_subgraphs: searchSubgraphs,
    recommend_subgraph: recommendSubgraph,
    get_subgraph_detail: getSubgraphDetail,
    list_registry_stats: listRegistryStats,
  };

• python/mcp_server.py:60-150 (handler)

  Python implementation of search_subgraphs tool. Queries SQLite database with same filtering logic as JavaScript version - filters by domain, network, protocol type, entity, reliability score, and keywords. Implements IPFS hash deduplication and returns JSON-formatted results.

  def search_subgraphs(
      query: str = "",
      domain: str = "",
      network: str = "",
      protocol_type: str = "",
      entity: str = "",
      min_reliability: float = 0.0,
      limit: int = 20,
  ) -> str:
      """Search and filter classified subgraphs."""
      conn = get_db()
      conditions = []
      params = []
  
      if domain:
          conditions.append("domain = ?")
          params.append(domain)
      if network:
          conditions.append("network = ?")
          params.append(network)
      if protocol_type:
          conditions.append("protocol_type = ?")
          params.append(protocol_type)
      if entity:
          conditions.append("canonical_entities LIKE ?")
          params.append(f'%"{entity}"%')
      if min_reliability > 0:
          conditions.append("reliability_score >= ?")
          params.append(min_reliability)
      if query:
          # Split into words and match any word (OR) across name/description/auto_description
          words = [w for w in query.strip().split() if len(w) > 2]
          if words:
              word_conds = []
              for w in words[:5]:
                  word_conds.append("(display_name LIKE ? OR description LIKE ? OR auto_description LIKE ?)")
                  params.extend([f"%{w}%", f"%{w}%", f"%{w}%"])
              conditions.append(f"({' OR '.join(word_conds)})")
          else:
              conditions.append("(display_name LIKE ? OR description LIKE ? OR auto_description LIKE ?)")
              params.extend([f"%{query}%", f"%{query}%", f"%{query}%"])
  
      where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
      # Over-fetch to allow dedup by IPFS hash (same deployment, different subgraph IDs)
      fetch_limit = limit * 3
      sql = f"""
          SELECT id, display_name, description, auto_description, domain, protocol_type, network,
                 reliability_score, ipfs_hash, entity_count, canonical_entities,
                 powered_by_substreams
          FROM subgraphs
          {where}
          ORDER BY reliability_score DESC
          LIMIT ?
      """
      params.append(fetch_limit)
  
      rows = conn.execute(sql, params).fetchall()
      conn.close()
  
      # Dedup by IPFS hash — keep highest reliability per deployment
      results = []
      seen_ipfs = set()
      for r in rows:
          ipfs = r["ipfs_hash"]
          if ipfs and ipfs in seen_ipfs:
              continue
          if ipfs:
              seen_ipfs.add(ipfs)
          results.append({
              "id": r["id"],
              "display_name": r["display_name"],
              "description": (r["description"] or r["auto_description"] or "")[:300],
              "domain": r["domain"],
              "protocol_type": r["protocol_type"],
              "network": r["network"],
              "reliability_score": r["reliability_score"],
              "ipfs_hash": r["ipfs_hash"],
              "entity_count": r["entity_count"],
              "canonical_entities": json.loads(r["canonical_entities"]),
              "powered_by_substreams": bool(r["powered_by_substreams"]),
              "query_url": f"https://gateway.thegraph.com/api/[api-key]/subgraphs/id/{r['id']}",
          })
          if len(results) >= limit:
              break
  
      return json.dumps({
          "total": len(results),
          "subgraphs": results,
          "query_instructions": "To query a subgraph: POST a GraphQL query to the query_url (replace [api-key] with your Graph API key from https://thegraph.com/studio/apikeys/). First fetch the schema with get_subgraph_detail to see available entities and fields.",
      }, indent=2)

• python/mcp_server.py:298-313 (schema)

  Python MCP tool schema definition for search_subgraphs with identical parameter structure to JavaScript version - defines input parameters with types and descriptions for filtering operations.

  {
      "name": "search_subgraphs",
      "description": "Search and filter the classified subgraph registry. Filter by domain (defi, nfts, dao, gaming, identity, infrastructure, social, analytics), network (mainnet, arbitrum-one, base, matic, bsc, optimism, avalanche), protocol_type (dex, lending, bridge, staking, options, perpetuals, nft-marketplace, yield-aggregator, governance, name-service), canonical entity type (liquidity_pool, trade, token, position, vault, loan, collateral, liquidation, nft_collection, nft_item, nft_sale, proposal, delegate, domain_name, account, transaction, daily_snapshot, hourly_snapshot), or free-text keyword. Returns subgraphs ranked by reliability score.",
      "inputSchema": {
          "type": "object",
          "properties": {
              "query": {"type": "string", "description": "Free-text search across names and descriptions"},
              "domain": {"type": "string", "description": "Filter by domain: defi, nfts, dao, gaming, identity, infrastructure, social, analytics"},
              "network": {"type": "string", "description": "Filter by chain: mainnet, arbitrum-one, base, matic, bsc, optimism, avalanche, etc."},
              "protocol_type": {"type": "string", "description": "Filter by protocol type: dex, lending, bridge, staking, options, perpetuals, etc."},
              "entity": {"type": "string", "description": "Filter by canonical entity: liquidity_pool, trade, token, position, vault, loan, etc."},
              "min_reliability": {"type": "number", "description": "Minimum reliability score (0-1). Higher = more signal/stake/fees."},
              "limit": {"type": "integer", "description": "Max results to return (default: 20)", "default": 20},
          },
      },
  },