Opinion.trade MCP Server

"""Pydantic validation models for Opinion.trade MCP server.""" from typing import Optional, Literal from pydantic import BaseModel, Field, model_validator # ============================================================================ # Public API Request Models (6 tools) # ============================================================================ class GetMarketsRequest(BaseModel): """Request model for get_markets tool.""" limit: int = Field(default=100, ge=1, le=500, description="Maximum markets to return") offset: int = Field(default=0, ge=0, description="Offset for pagination") status: Optional[str] = Field(default=None, description="Filter by status (active/closed/all)") class GetMarketDetailsRequest(BaseModel): """Request model for get_market_details tool.""" market_id: str = Field(..., min_length=1, description="Market ID to retrieve") class GetTokenPriceRequest(BaseModel): """Request model for get_token_price tool.""" token_id: str = Field(..., min_length=1, description="Token ID for price lookup") class GetOrderbookRequest(BaseModel): """Request model for get_orderbook tool.""" token_id: str = Field(..., min_length=1, description="Token ID for orderbook") depth: int = Field(default=20, ge=1, le=100, description="Number of price levels") class GetPriceHistoryRequest(BaseModel): """Request model for get_price_history tool.""" token_id: str = Field(..., min_length=1, description="Token ID for price history") timeframe: Literal["1m", "5m", "15m", "1h", "4h", "1d", "7d", "30d"] = Field( default="1h", description="Timeframe for OHLCV data" ) limit: int = Field(default=100, ge=1, le=1000, description="Number of data points") class SearchMarketsRequest(BaseModel): """Request model for search_markets tool.""" query: str = Field(..., min_length=1, description="Search query") limit: int = Field(default=20, ge=1, le=100, description="Maximum results") # ============================================================================ # Trading Request Models (7 tools) # ============================================================================ class PlaceOrderRequest(BaseModel): """Request model for place_order tool.""" token_id: str = Field(..., min_length=1, description="Token ID to trade") side: Literal["BUY", "SELL"] = Field(..., description="Order side") amount: float = Field(..., gt=0, description="Order amount/size") price: Optional[float] = Field(None, gt=0, le=1, description="Limit price (0-1)") order_type: Literal["LIMIT", "MARKET"] = Field(default="LIMIT", description="Order type") @model_validator(mode="after") def validate_price_for_limit(self): """Validate that LIMIT orders have a price.""" if self.order_type == "LIMIT" and self.price is None: raise ValueError("Price is required for LIMIT orders") if self.order_type == "MARKET" and self.price is not None: raise ValueError("Price should not be provided for MARKET orders") return self class CancelOrderRequest(BaseModel): """Request model for cancel_order tool.""" order_id: str = Field(..., min_length=1, description="Order ID to cancel") class CancelAllOrdersRequest(BaseModel): """Request model for cancel_all_orders tool.""" market_id: Optional[str] = Field(None, description="Optional market filter") class GetOpenOrdersRequest(BaseModel): """Request model for get_open_orders tool.""" market_id: Optional[str] = Field(None, description="Optional market filter") class GetPositionsRequest(BaseModel): """Request model for get_positions tool.""" market_id: Optional[str] = Field(None, description="Optional market filter") class GetTradeHistoryRequest(BaseModel): """Request model for get_trade_history tool.""" market_id: Optional[str] = Field(None, description="Optional market filter") limit: int = Field(default=100, ge=1, le=1000, description="Maximum trades") start_time: Optional[int] = Field(None, description="Start timestamp (ms)") end_time: Optional[int] = Field(None, description="End timestamp (ms)") class GetBalancesRequest(BaseModel): """Request model for get_balances tool.""" pass # No parameters needed # ============================================================================ # Error Response Model # ============================================================================ class APIError(BaseModel): """Standardized error response model.""" error: str = Field(..., description="Error message") error_code: Optional[str] = Field(None, description="Error code identifier") status_code: Optional[int] = Field(None, description="HTTP status code") is_error: bool = Field(default=True, description="Error flag") def model_dump_json(self, **kwargs) -> str: """Override to ensure consistent JSON serialization.""" return super().model_dump_json(indent=2, **kwargs)