uniswap-pools-mcp

import os from mcp.server.fastmcp import FastMCP, Context from typing import List, Dict import httpx import json from dataclasses import dataclass from dotenv import load_dotenv import pandas as pd # Load environment variables from .env file load_dotenv() # Initialize MCP server mcp = FastMCP("Uniswap Pools", dependencies=["httpx", "python-dotenv", "pandas"]) # Retrieve API key from .env API_KEY = os.getenv("THEGRAPH_API_KEY") if not API_KEY: raise ValueError("THEGRAPH_API_KEY not found in .env file") BASE_URL = "https://gateway.thegraph.com/api/subgraphs/id" # Uniswap V2 Subgraph endpoint UNISWAP_V2_SUBGRAPH = f"{BASE_URL}/A3Np3RQbaBA6oKJgiwDJeo5T3zrYfGHPWFYayMwtNDum" # Uniswap V3 Subgraph endpoint UNISWAP_V3_SUBGRAPH = f"{BASE_URL}/5zvR82QoaXYFyDEKLZ9t6v9adgnptxYpKpSbxtgVENFV" # Uniswap V4 Subgraph endpoint UNISWAP_V4_SUBGRAPH = f"{BASE_URL}/DiYPVdygkfjDWhbxGSqAQxwBKmfKnkWQojqeM2rkLb3G" # Define data structure for V2 pair @dataclass class Pair: id: str token0: str token1: str reserveUSD: str volumeUSD: str pair: str # Format: token0.symbol/token1.symbol # Define data structure for V3 pool @dataclass class Pool: id: str token0: str token1: str feeTier: int liquidity: str volumeUSD: str feesUSD: str pair: str # Format: token0.symbol/token1.symbol # Define data structure for V4 pool @dataclass class PoolV4: id: str token0: str token1: str feeTier: str liquidity: str volumeUSD: str feesUSD: str pair: str # Format: token0.symbol/token1.symbol # Query Subgraph to fetch V2 pairs for a given token async def query_pairs_v2(token_address: str) -> List[Pair]: query = """ query($token: ID!) { pairs( where: { or: [ {token0: $token}, {token1: $token} ] } orderBy: volumeUSD orderDirection: desc ) { id token0 { id symbol } token1 { id symbol } reserveUSD volumeUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V2_SUBGRAPH, headers={ "Authorization": f"Bearer {API_KEY}" }, json={ "query": query, "variables": {"token": token_address.lower()} } ) response.raise_for_status() data = response.json() if "errors" in data: raise ValueError(f"GraphQL errors: {data['errors']}") return [ Pair( id=pair["id"], token0=pair["token0"]["id"], token1=pair["token1"]["id"], reserveUSD=pair["reserveUSD"], volumeUSD=pair["volumeUSD"], pair=f"{pair['token0']['symbol']}/{pair['token1']['symbol']}" ) for pair in data["data"]["pairs"] ] # Query Subgraph to fetch a specific V2 pair by ID async def query_pair_v2_by_id(pair_id: str) -> Pair: query = """ query($id: ID!) { pairs(where: { id: $id }) { id token0 { id symbol } token1 { id symbol } reserveUSD volumeUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V2_SUBGRAPH, headers={ "Authorization": f"Bearer {API_KEY}" }, json={ "query": query, "variables": {"id": pair_id.lower()} } ) response.raise_for_status() data = response.json() if "errors" in data: raise ValueError(f"GraphQL errors: {data['errors']}") pairs = data["data"]["pairs"] if not pairs: raise ValueError(f"No V2 pair found with ID: {pair_id}") pair = pairs[0] return Pair( id=pair["id"], token0=pair["token0"]["id"], token1=pair["token1"]["id"], reserveUSD=pair["reserveUSD"], volumeUSD=pair["volumeUSD"], pair=f"{pair['token0']['symbol']}/{pair['token1']['symbol']}" ) # Query Subgraph to fetch V3 pools for a given token async def query_pools_v3(token_address: str) -> List[Pool]: query = """ query($token: Bytes!) { pools( where: { or: [ {token0: $token}, {token1: $token} ] } orderBy: volumeUSD orderDirection: desc ) { id token0 { id symbol } token1 { id symbol } feeTier liquidity volumeUSD feesUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V3_SUBGRAPH, headers={ "Authorization": f"Bearer {API_KEY}" }, json={ "query": query, "variables": {"token": token_address.lower()} } ) response.raise_for_status() data = response.json() if "errors" in data: raise ValueError(f"GraphQL errors: {data['errors']}") return [ Pool( id=pool["id"], token0=pool["token0"]["id"], token1=pool["token1"]["id"], feeTier=pool["feeTier"], liquidity=pool["liquidity"], volumeUSD=pool["volumeUSD"], feesUSD=pool["feesUSD"], pair=f"{pool['token0']['symbol']}/{pool['token1']['symbol']}" ) for pool in data["data"]["pools"] ] # Query Subgraph to fetch a specific V3 pool by ID async def query_pool_v3_by_id(pool_id: str) -> Pool: query = """ query($id: ID!) { pools(where: { id: $id }) { id token0 { id symbol } token1 { id symbol } feeTier liquidity volumeUSD feesUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V3_SUBGRAPH, headers={ "Authorization": f"Bearer {API_KEY}" }, json={ "query": query, "variables": {"id": pool_id.lower()} } ) response.raise_for_status() data = response.json() if "errors" in data: raise ValueError(f"GraphQL errors: {data['errors']}") pools = data["data"]["pools"] if not pools: raise ValueError(f"No V3 pool found with ID: {pool_id}") pool = pools[0] return Pool( id=pool["id"], token0=pool["token0"]["id"], token1=pool["token1"]["id"], feeTier=pool["feeTier"], liquidity=pool["liquidity"], volumeUSD=pool["volumeUSD"], feesUSD=pool["feesUSD"], pair=f"{pool['token0']['symbol']}/{pool['token1']['symbol']}" ) # Query Subgraph to fetch V4 pools for a given token async def query_pools_v4(token_address: str) -> List[PoolV4]: query = """ query($token: ID!) { pools( where: { or: [ {token0: $token}, {token1: $token} ] } orderBy: volumeUSD orderDirection: desc ) { id token0: token0 { id symbol } token1: token1 { id symbol } feeTier liquidity volumeUSD feesUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V4_SUBGRAPH, headers={ 'Authorization': f'Bearer {API_KEY}', }, json={ 'query': query, 'variables': {'token': token_address.lower()} } ) response.raise_for_status() data = response.json() if 'errors' in data: raise ValueError(f"GraphQL errors: {data['errors']}") return [ PoolV4( id=pool['id'], token0=pool['token0']['id'], token1=pool['token1']['id'], feeTier=str(pool['feeTier']), liquidity=str(pool['liquidity']), volumeUSD=pool['volumeUSD'], feesUSD=pool['feesUSD'], pair=f"{pool['token0']['symbol']}/{pool['token1']['symbol']}" ) for pool in data['data']['pools'] ] # Query Subgraph to fetch a specific V4 pool by ID async def query_pool_v4_by_id(pool_id: str) -> PoolV4: query = """ query($id: ID!) { pools(where: { id: $id }) { id token0 { id symbol } token1 { id symbol } feeTier liquidity volumeUSD feesUSD } } """ async with httpx.AsyncClient() as client: response = await client.post( UNISWAP_V4_SUBGRAPH, headers={ "Authorization": f"Bearer {API_KEY}" }, json={ "query": query, "variables": {"id": pool_id.lower()} } ) response.raise_for_status() data = response.json() if "errors" in data: raise ValueError(f"GraphQL errors: {data['errors']}") pools = data["data"]["pools"] if not pools: raise ValueError(f"No V4 pool found with ID: {pool_id}") pool = pools[0] return PoolV4( id=pool["id"], token0=pool["token0"]["id"], token1=pool["token1"]["id"], feeTier=str(pool["feeTier"]), liquidity=str(pool["liquidity"]), volumeUSD=pool["volumeUSD"], feesUSD=pool["feesUSD"], pair=f"{pool['token0']['symbol']}/{pool['token1']['symbol']}" ) # Tool: Query all Uniswap V2, V3, and V4 pools/pairs for a specific token and return as a formatted table @mcp.tool() async def get_token_pools(token_address: str, ctx: Context) -> str: """ Query all Uniswap V2, V3, and V4 pools/pairs for a specific token and return as a formatted markdown table. Parameters: token_address (str): The Ethereum address of the token to query (e.g., '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'). ctx (Context): The API context for logging and error handling. Returns: A markdown-formatted string containing a table with columns: Version, ID, Pair, Fee Tier, Volume USD, Liquidity/ReserveUSD, Fees USD, sorted by Volume USD in descending order. """ ctx.info(f"Querying V2, V3, and V4 pools/pairs for token: {token_address}") try: # Query V2 pairs v2_pairs = await query_pairs_v2(token_address) ctx.info(f"Found {len(v2_pairs)} V2 pairs") # Query V3 pools v3_pools = await query_pools_v3(token_address) ctx.info(f"Found {len(v3_pools)} V3 pools") # Query V4 pools v4_pools = await query_pools_v4(token_address) ctx.info(f"Found {len(v4_pools)} V4 pools") # Combine V2, V3, and V4 data into a single DataFrame df = pd.DataFrame([ { "Version": "v2", "ID": pair.id, "Pair": pair.pair, "Fee Tier": "3000", # V2 fixed 0.3% fee (3000 basis points) "Volume USD": pair.volumeUSD, "Liquidity/ReserveUSD": pair.reserveUSD, "Fees USD": "N/A" # V2 schema doesn't provide feesUSD } for pair in v2_pairs ] + [ { "Version": "v3", "ID": pool.id, "Pair": pool.pair, "Fee Tier": str(pool.feeTier), "Volume USD": pool.volumeUSD, "Liquidity/ReserveUSD": pool.liquidity, "Fees USD": pool.feesUSD } for pool in v3_pools ] + [ { "Version": "v4", "ID": pool.id, "Pair": pool.pair, "Fee Tier": pool.feeTier, "Volume USD": pool.volumeUSD, "Liquidity/ReserveUSD": pool.liquidity, "Fees USD": pool.feesUSD } for pool in v4_pools ]) # Sort by Volume USD (descending) and select columns df = df.sort_values(by="Volume USD", ascending=False) return df.to_markdown(index=False) except Exception as e: ctx.error(f"Failed to query V2/V3/V4 pools: {str(e)}") raise # Tool: Query all Uniswap V2 pairs for a specific token and return as a formatted table @mcp.tool() async def get_token_pools_v2(token_address: str, ctx: Context) -> str: """ Query all Uniswap V2 pairs for a specific token and return as a formatted markdown table. Parameters: token_address (str): The Ethereum address of the token to query (e.g., '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'). ctx (Context): The API context for logging and error handling. Returns: A markdown-formatted string containing a table with columns: Version, ID, Pair, Volume USD, ReserveUSD. """ ctx.info(f"Querying V2 pairs for token: {token_address}") try: pairs = await query_pairs_v2(token_address) ctx.info(f"Found {len(pairs)} V2 pairs") # Create DataFrame directly from pairs list df = pd.DataFrame([ { "Version": "v2", "ID": pair.id, "Pair": pair.pair, "Volume USD": pair.volumeUSD, "ReserveUSD": pair.reserveUSD } for pair in pairs ]) return df.to_markdown(index=False) except Exception as e: ctx.error(f"Failed to query V2 pairs: {str(e)}") raise # Tool: Query all Uniswap V3 pools for a specific token and return as a formatted table @mcp.tool() async def get_token_pools_v3(token_address: str, ctx: Context) -> str: """ Query all Uniswap V3 pools for a specific token and return as a formatted markdown table. Parameters: token_address (str): The Ethereum address of the token to query (e.g., '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'). ctx (Context): The API context for logging and error handling. Returns: A markdown-formatted string containing a table with columns: Version, ID, Pair, Fee Tier, Volume USD, Liquidity. """ ctx.info(f"Querying V3 pools for token: {token_address}") try: pools = await query_pools_v3(token_address) ctx.info(f"Found {len(pools)} V3 pools") # Create DataFrame directly from pools list df = pd.DataFrame([ { "Version": "v3", "ID": pool.id, "Pair": pool.pair, "Fee Tier": pool.feeTier, "Volume USD": pool.volumeUSD, "Liquidity": pool.liquidity } for pool in pools ]) return df.to_markdown(index=False) except Exception as e: ctx.error(f"Failed to query V3 pools: {str(e)}") raise # Tool: Query all Uniswap V4 pools for a specific token and return as a formatted table @mcp.tool() async def get_token_pools_v4(token_address: str, ctx: Context) -> str: """ Query all Uniswap V4 pools for a specific token and return as a formatted markdown table. Parameters: token_address (str): The Ethereum address of the token to query (e.g., '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'). ctx (Context): The API context for logging and error handling. Returns: A markdown-formatted string containing a table with columns: Version, ID, Pair, Fee Tier, Volume USD, Liquidity. """ ctx.info(f"Querying V4 pools for token: {token_address}") try: pools = await query_pools_v4(token_address) ctx.info(f"Found {len(pools)} V4 pools") # Create DataFrame directly from pools list df = pd.DataFrame([ { "Version": "v4", "ID": pool.id, "Pair": pool.pair, "Fee Tier": pool.feeTier, "Volume USD": pool.volumeUSD, "Liquidity": pool.liquidity } for pool in pools ]) return df.to_markdown(index=False) except Exception as e: ctx.error(f"Failed to query V4 pools: {str(e)}") raise # Tool: Query a specific Uniswap pool/pair by version and ID and return as markdown text @mcp.tool() async def get_pool_data(version: str, pool_id: str, ctx: Context) -> str: """ Query a specific Uniswap pool/pair by version (v2, v3, v4) and ID and return as markdown text. Parameters: version (str): The Uniswap version to query ('v2', 'v3', or 'v4'). pool_id (str): The Ethereum address of the pool or pair to query (e.g., '0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc'). ctx (Context): The API context for logging and error handling. Returns: A markdown-formatted string containing details of the pool/pair, including Version, ID, Pair, Token0 Address, Token1 Address, Fee Tier, Volume USD, Liquidity/ReserveUSD, and Fees USD. """ ctx.info(f"Querying {version} pool/pair with ID: {pool_id}") try: if version.lower() == "v2": pair = await query_pair_v2_by_id(pool_id) result = f""" **Uniswap V2 Pair Details** - **Version**: v2 - **ID**: {pair.id} - **Pair**: {pair.pair} - **Token0 Address**: {pair.token0} - **Token1 Address**: {pair.token1} - **Fee Tier**: 3000 (0.3%) - **Volume USD**: {pair.volumeUSD} - **Liquidity/ReserveUSD**: {pair.reserveUSD} - **Fees USD**: N/A """ elif version.lower() == "v3": pool = await query_pool_v3_by_id(pool_id) result = f""" **Uniswap V3 Pool Details** - **Version**: v3 - **ID**: {pool.id} - **Pair**: {pool.pair} - **Token0 Address**: {pool.token0} - **Token1 Address**: {pool.token1} - **Fee Tier**: {pool.feeTier} - **Volume USD**: {pool.volumeUSD} - **Liquidity/ReserveUSD**: {pool.liquidity} - **Fees USD**: {pool.feesUSD} """ elif version.lower() == "v4": pool = await query_pool_v4_by_id(pool_id) result = f""" **Uniswap V4 Pool Details** - **Version**: v4 - **ID**: {pool.id} - **Pair**: {pool.pair} - **Token0 Address**: {pool.token0} - **Token1 Address**: {pool.token1} - **Fee Tier**: {pool.feeTier} - **Volume USD**: {pool.volumeUSD} - **Liquidity/ReserveUSD**: {pool.liquidity} - **Fees USD**: {pool.feesUSD} """ else: raise ValueError(f"Invalid version: {version}. Must be 'v2', 'v3', or 'v4'") return result.strip() except Exception as e: ctx.error(f"Failed to query {version} pool/pair: {str(e)}") raise # Main entry point if __name__ == "__main__": mcp.run()