PancakeSwap PoolSpy MCP Server

Instructions

Implementation Reference

• main.py:68-98 (handler)

  The main tool handler for 'get_new_pools_bsc', registered via @mcp.tool() decorator. Fetches recent pools using subgraph query and formats output as a string.

  @mcp.tool()
  async def get_new_pools_bsc(time_range_seconds: int = 300, limit: int = 100) -> str:
      """
      Returns a list of trading pools created in the specified time range on Pancake Swap V3 BNB Smart Chain.
  
      Parameters:
          time_range_seconds (int): The time range in seconds to look back for new pools.
                                    Default is 300 seconds (5 minutes).
          limit (int): The maximum number of pools to return.
                       Default is 100 pools.
      """
      try:
          pools = await fetch_recent_pools(time_range_seconds=time_range_seconds, limit=limit)
          time_range_minutes = time_range_seconds // 60  # Convert to minutes for display
          output = f"Newly Created Trading Pools (Last {time_range_minutes} Minutes, Limit: {limit}):\n"
          for pool in pools:
              timestamp = datetime.fromtimestamp(int(pool["createdAtTimestamp"])).strftime('%Y-%m-%d %H:%M:%S')
              volume_usd = float(pool["volumeUSD"])  # Ensure float for formatting
              tvl_usd = float(pool["totalValueLockedUSD"])  # Ensure float for formatting
              output += (
                  f"Pool Address: {pool['id']}\n"
                  f"Tokens: {pool['token0']['symbol']}/{pool['token1']['symbol']}\n"
                  f"Created At: {timestamp}\n"
                  f"Block Number: {pool['createdAtBlockNumber']}\n"
                  f"Transaction Count: {pool['txCount']}\n"
                  f"Volume (USD): {volume_usd:.2f}\n"
                  f"Total Value Locked (USD): {tvl_usd:.2f}\n\n"
              )
          return output if pools else f"No pools created in the last {time_range_minutes} minutes."
      except Exception as e:
          return f"Error fetching new pools: {str(e)}"

• main.py:35-65 (helper)

  Core helper function that constructs and executes the GraphQL query to retrieve recently created pools from the PancakeSwap V3 subgraph.

  async def fetch_recent_pools(time_range_seconds: int = 300, limit: int = 100) -> list[dict]:
      time_ago = int((datetime.utcnow() - timedelta(seconds=time_range_seconds)).timestamp())
      query = """
      query RecentPools($timestamp: BigInt!, $limit: Int!) {
          pools(
              where: { createdAtTimestamp_gt: $timestamp }
              orderBy: createdAtTimestamp
              orderDirection: desc
              first: $limit
          ) {
              id
              token0 { symbol }
              token1 { symbol }
              createdAtTimestamp
              createdAtBlockNumber
              txCount
              volumeUSD
              totalValueLockedUSD
          }
      }
      """
      variables = {"timestamp": str(time_ago), "limit": limit}  # Convert timestamp to string for BigInt
      try:
          result = await query_subgraph(query, variables)
          pools = result.get("data", {}).get("pools", [])
          if not pools:
              print(f"No pools found for timestamp > {time_ago}. Response: {json.dumps(result, indent=2)}")
          return pools
      except Exception as e:
          print(f"Error in fetch_recent_pools: {str(e)}")
          raise

• main.py:21-32 (helper)

  Utility helper for executing GraphQL queries against the configured PancakeSwap subgraph endpoint.

  async def query_subgraph(query: str, variables: dict = None) -> dict:
      async with httpx.AsyncClient(timeout=10.0) as client:
          payload = {"query": query}
          if variables:
              payload["variables"] = variables
          response = await client.post(SUBGRAPH_URL, json=payload)
          if response.status_code != 200:
              raise Exception(f"Subgraph query failed with status {response.status_code}: {response.text}")
          result = response.json()
          if "errors" in result:
              raise Exception(f"GraphQL errors: {json.dumps(result['errors'], indent=2)}")
          return result

• main.py:68-68 (registration)

  The @mcp.tool() decorator registers the get_new_pools_bsc function as an MCP tool.

  @mcp.tool()