get_token_pools_v2
Retrieve Uniswap V2 token pairs by token address and generate a markdown-formatted table with details like pair ID, volume, and reserves for easy analysis and integration.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| token_address | Yes |
Implementation Reference
- main.py:481-513 (handler)The handler function for the 'get_token_pools_v2' tool, registered via @mcp.tool(). It queries Uniswap V2 pairs for the specified token address using the helper function query_pairs_v2, constructs a pandas DataFrame, and returns a markdown-formatted table of results.@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
- main.py:67-123 (helper)Supporting helper function that performs the GraphQL query to the Uniswap V2 subgraph to fetch all pairs containing the given token, ordered by volumeUSD descending, and parses the response into a list of Pair dataclasses.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"] ]
- main.py:33-41 (schema)Dataclass schema defining the structure of Uniswap V2 pair data used by the handler and helper.@dataclass class Pair: id: str token0: str token1: str reserveUSD: str volumeUSD: str pair: str # Format: token0.symbol/token1.symbol