PancakeSwap PoolSpy MCP Server

Instructions

Implementation Reference

• main.py:69-98 (handler)

  The main asynchronous handler function that implements the core logic of the 'get_new_pools_bsc' tool. It calls the helper function to fetch recent pools from the PancakeSwap V3 subgraph and formats the results into a human-readable string output.
  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:68-68 (registration)

  The @mcp.tool() decorator registers the 'get_new_pools_bsc' function as an MCP tool with FastMCP.
  @mcp.tool()
• main.py:35-65 (helper)

  Helper function that constructs and executes the GraphQL query to fetch recently created pools from the subgraph within the specified time range.
  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 function to perform asynchronous HTTP POST requests to the GraphQL 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:69-78 (schema)

  Input schema defined by function parameters with type hints and defaults, output as str, detailed in docstring.
  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. """