MCP Paradex Server

Instructions

Implementation Reference

• src/mcp_paradex/tools/vaults.py:37-117 (handler)

  The primary handler function for the 'paradex_vaults' tool. It fetches vault data from the Paradex API, validates it using the Vault Pydantic model via TypeAdapter, applies optional JMESPath filtering, sorting by creation date descending, pagination with limit/offset, and returns a structured response including the schema and metadata.
  @server.tool(name="paradex_vaults") async def get_vaults( vault_address: Annotated[ str, Field( default="", description="The address of the vault to get details for or empty string to get all vaults.", ), ], jmespath_filter: Annotated[ str, Field( default=None, description="JMESPath expression to filter, sort, or limit the results." ), ], 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.", ), ], ) -> dict: """ Get detailed information about a specific vault or all vaults if no address is provided. Retrieves comprehensive details about a specific vault identified by its address, including configuration, permissions, and other vault-specific parameters. Use jmespath_filter to reduce the number of results as much as possible as number of vaults can be large. You can use JMESPath expressions to filter, sort, or limit the results. Examples: - Filter by owner: "[?owner_account=='0x123...']" - Filter by status: "[?status=='ACTIVE']" - Find vaults with specific strategy: "[?contains(strategies, 'strategy_id')]" - Sort by creation date: "sort_by([*], &created_at)" - Limit to newest vaults: "sort_by([*], &created_at)[-5:]" - Select specific fields: "[*].{address: address, name: name, kind: kind, status: status}" """ try: client = await get_paradex_client() params = {"address": vault_address} if vault_address else None response = await api_call(client, "vaults", params=params) if "error" in response: raise Exception(response["error"]) results = response["results"] vaults = vault_adapter.validate_python(results) # Apply JMESPath filter if provided if jmespath_filter: vaults = apply_jmespath_filter( data=vaults, jmespath_filter=jmespath_filter, type_adapter=vault_adapter, error_logger=logger.error, ) sorted_vaults = sorted(vaults, key=lambda x: x.created_at, reverse=True) result_vaults = sorted_vaults[offset : offset + limit] result = { "description": Vault.__doc__.strip() if Vault.__doc__ else None, "fields": Vault.model_json_schema(), "vaults": result_vaults, "total": len(sorted_vaults), "limit": limit, "offset": offset, } return result except Exception as e: logger.error(f"Error fetching vault details: {e!s}") raise e
• src/mcp_paradex/models.py:219-263 (schema)

  Pydantic BaseModel 'Vault' that defines the schema for individual vault objects. Used for input validation, output typing, and schema generation (Vault.model_json_schema()). Referenced in the handler's return value and validation.
  class Vault(BaseModel): """Vault model representing a trading vault on Paradex.""" address: Annotated[str, Field(default="", description="Contract address of the vault")] name: Annotated[str, Field(default="", description="Name of the vault")] description: Annotated[str, Field(default="", description="Description of the vault")] owner_account: Annotated[str, Field(default="", description="Owner account of the vault")] operator_account: Annotated[str, Field(default="", description="Operator account of the vault")] strategies: Annotated[ list[str], Field(default_factory=list, description="Strategies of the vault") ] token_address: Annotated[str, Field(default="", description="LP token address")] status: Annotated[str, Field(default="", description="Status of the vault")] kind: Annotated[ str, Field( default="", description="Kind of the vault: 'user' for user-defined vaults, 'protocol' for vaults controlled by Paradex", ), ] profit_share: Annotated[ int, Field(default=0, description="Profit share of the vault in percentage, i.e. 10 means 10%"), ] lockup_period: Annotated[ int, Field(default=0, description="Lockup period of the vault in days") ] max_tvl: Annotated[ int, Field(default=0, description="Maximum amount of assets the vault can hold in USDC") ] created_at: Annotated[ int, Field( default=0, description="Unix timestamp in milliseconds of when the vault has been created", ), ] last_updated_at: Annotated[ int, Field( default=0, description="Unix timestamp in milliseconds of when the vault was last updated", ), ]
• src/mcp_paradex/tools/market.py:47-56 (registration)

  The 'paradex_vaults' tool schema is registered in the get_filters_model tool's description dictionary, providing Vault.model_json_schema() for client-side filtering schema information.
  tool_descriptions = { "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/vaults.py:34-35 (helper)

  TypeAdapter for list[Vault] used in the handler to validate API response data before processing.
  vault_adapter = TypeAdapter(list[Vault])