MCP Paradex Server

Instructions

Implementation Reference

• src/mcp_paradex/tools/market.py:158-251 (handler)

  The primary handler function for the 'paradex_market_summaries' tool. It fetches market summary data from the Paradex client, validates it using the MarketSummary Pydantic model, applies optional JMESPath filtering, pagination, and returns a structured response with metadata.
  @server.tool(name="paradex_market_summaries") async def get_market_summaries( market_ids: Annotated[ list[str], Field(description="Market symbols to get summaries for.", default=["ALL"]) ], jmespath_filter: Annotated[ str, Field( description="JMESPath expression to filter, sort, or limit the results.", default=None, ), ], limit: Annotated[ int, Field( default=10, gt=0, le=100, description="Limit the number of results to the specified number.", ), ], offset: Annotated[ int, Field( default=0, ge=0, description="Offset the results to the specified number.", ), ], ctx: Context = None, ) -> dict: """ Identify the most active or volatile markets and get current market conditions. Use this tool when you need to: - Find the most active markets by volume for liquidity analysis - Discover markets with significant price movements for momentum strategies - Compare current prices across multiple assets - Identify markets with unusual behavior for potential opportunities Retrieves current market summary information including price, volume, 24h change, and other key market metrics. If "ALL" is specified or no market IDs are provided, returns summaries for all available markets. Example use cases: - Finding high-volatility markets for short-term trading - Identifying top gainers and losers for the day - Comparing volume across different markets to find liquidity - Getting the current price and 24-hour range for price analysis You can use JMESPath expressions (https://jmespath.org/specification.html) to filter, sort, or limit the results. Use the `paradex_filters_model` tool to get the filters for a tool. Examples: - Filter by high price: "[?high_price > `10000`]" - Sort by volume: "sort_by([*], &volume)" - Get top 3 by price change: "[sort_by([*], &to_number(price_change_percent))[-3:]]" """ try: # Get market summary from Paradex client = await get_paradex_client() response = client.fetch_markets_summary(params={"market": "ALL"}) if "error" in response: await ctx.error(response) raise Exception(response["error"]) # Try to validate directly now that the model is more flexible summaries = market_summary_adapter.validate_python(response["results"]) if market_ids and "ALL" not in market_ids: summaries = [summary for summary in summaries if summary.symbol in market_ids] # Apply JMESPath filter if provided if jmespath_filter: summaries = apply_jmespath_filter( data=summaries, jmespath_filter=jmespath_filter, type_adapter=market_summary_adapter, error_logger=ctx.error if ctx else None, ) sorted_summaries = sorted(summaries, key=lambda x: x.symbol, reverse=True) result_summaries = sorted_summaries[offset : offset + limit] result = { "description": MarketSummary.__doc__.strip() if MarketSummary.__doc__ else None, "fields": MarketSummary.model_json_schema(), "results": result_summaries, "total": len(sorted_summaries), "limit": limit, "offset": offset, } return result except Exception as e: logger.error(f"Error fetching market summaries: {e!s}") await ctx.error(f"Error fetching market summaries: {e!s}") raise e
• src/mcp_paradex/models.py:416-444 (schema)

  Pydantic model defining the schema and validation rules for individual market summary objects returned by the tool.
  class MarketSummary(BaseModel): """Model representing a summary of a market.""" symbol: Annotated[str, Field(default="", description="Market symbol")] mark_price: Annotated[str, Field(default="", description="Mark price")] delta: Annotated[str, Field(default="", description="Deprecated: Use greeks.delta instead")] greeks: Annotated[ Greeks, Field( default=None, description="Greeks (delta, gamma, vega). Partial for perpetual futures" ), ] last_traded_price: Annotated[str, Field(default="", description="Last traded price")] bid: Annotated[str, Field(default="", description="Best bid price")] ask: Annotated[str, Field(default="", description="Best ask price")] volume_24h: Annotated[str, Field(default="", description="24 hour volume in USD")] total_volume: Annotated[ str, Field(default="", description="Lifetime total traded volume in USD") ] created_at: Annotated[int, Field(default=0, description="Market summary creation time")] underlying_price: Annotated[ str, Field(default="", description="Underlying asset price (spot price)") ] open_interest: Annotated[str, Field(default="", description="Open interest in base currency")] funding_rate: Annotated[str, Field(default="", description="8 hour funding rate")] price_change_rate_24h: Annotated[ str, Field(default="", description="Price change rate in the last 24 hours") ]
• src/mcp_paradex/tools/market.py:158-158 (registration)

  The @server.tool decorator registers the get_market_summaries function as the MCP tool named 'paradex_market_summaries'.
  @server.tool(name="paradex_market_summaries")
• src/mcp_paradex/tools/market.py:48-55 (schema)

  Entry in the paradex_filters_model tool that provides the JSON schema for paradex_market_summaries input/output via MarketSummary.model_json_schema().
  "paradex_markets": models.MarketDetails.model_json_schema(), "paradex_market_summaries": models.MarketSummary.model_json_schema(), "paradex_open_orders": models.OrderState.model_json_schema(), "paradex_orders_history": models.OrderState.model_json_schema(), "paradex_vaults": models.Vault.model_json_schema(), "paradex_vault_summary": models.VaultSummary.model_json_schema(), } return tool_descriptions[tool_name]
• src/mcp_paradex/tools/market.py:155-155 (helper)

  TypeAdapter for list[MarketSummary] used for Pydantic validation of the API response data.
  market_summary_adapter = TypeAdapter(list[MarketSummary])