Trade Me MCP Server

"""Trade Me API client.""" import os import logging from typing import Any, Optional import httpx from .auth import TradeMeAuth logger = logging.getLogger(__name__) class TradeMeClient: """Client for interacting with the Trade Me API.""" SANDBOX_URL = "https://api.tmsandbox.co.nz/v1" PRODUCTION_URL = "https://api.trademe.co.nz/v1" def __init__( self, auth: Optional[TradeMeAuth] = None, environment: str = "sandbox", timeout: float = 30.0, ): """ Initialize Trade Me API client. Args: auth: TradeMeAuth instance (creates one if not provided) environment: 'sandbox' or 'production' timeout: Request timeout in seconds """ self.auth = auth or TradeMeAuth() self.environment = environment or os.getenv("TRADEME_ENVIRONMENT", "sandbox") self.timeout = timeout self.base_url = ( self.SANDBOX_URL if self.environment == "sandbox" else self.PRODUCTION_URL ) self.client = httpx.Client( base_url=self.base_url, timeout=self.timeout, auth=self.auth.auth, headers={ "Accept": "application/json", "User-Agent": "TradeMeMCPServer/0.1.0", }, ) logger.info(f"Initialized Trade Me client for {self.environment} environment") def _make_request( self, method: str, endpoint: str, params: Optional[dict[str, Any]] = None, json_data: Optional[dict[str, Any]] = None, requires_auth: bool = False, ) -> dict[str, Any]: """ Make an HTTP request to the Trade Me API. Args: method: HTTP method (GET, POST, etc.) endpoint: API endpoint (without base URL) params: Query parameters json_data: JSON body data requires_auth: Whether this endpoint requires authentication Returns: API response as dictionary Raises: httpx.HTTPError: If the request fails ValueError: If authentication is required but not available """ if requires_auth and not self.auth.is_authenticated(): raise ValueError( "This endpoint requires authentication. " "Please set TRADEME_ACCESS_TOKEN and TRADEME_ACCESS_TOKEN_SECRET." ) # Remove leading slash if present endpoint = endpoint.lstrip("/") try: logger.debug(f"{method} {endpoint} - params: {params}") response = self.client.request( method=method, url=endpoint, params=params, json=json_data, ) response.raise_for_status() # Handle empty responses if response.status_code == 204 or not response.content: return {} return response.json() except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}") raise except httpx.RequestError as e: logger.error(f"Request error: {str(e)}") raise except Exception as e: logger.error(f"Unexpected error: {str(e)}") raise def get( self, endpoint: str, params: Optional[dict[str, Any]] = None, requires_auth: bool = False, ) -> dict[str, Any]: """Make a GET request.""" return self._make_request("GET", endpoint, params=params, requires_auth=requires_auth) def post( self, endpoint: str, json_data: Optional[dict[str, Any]] = None, params: Optional[dict[str, Any]] = None, requires_auth: bool = True, ) -> dict[str, Any]: """Make a POST request.""" return self._make_request( "POST", endpoint, params=params, json_data=json_data, requires_auth=requires_auth ) def put( self, endpoint: str, json_data: Optional[dict[str, Any]] = None, params: Optional[dict[str, Any]] = None, requires_auth: bool = True, ) -> dict[str, Any]: """Make a PUT request.""" return self._make_request( "PUT", endpoint, params=params, json_data=json_data, requires_auth=requires_auth ) def delete( self, endpoint: str, params: Optional[dict[str, Any]] = None, requires_auth: bool = True, ) -> dict[str, Any]: """Make a DELETE request.""" return self._make_request("DELETE", endpoint, params=params, requires_auth=requires_auth) def close(self) -> None: """Close the HTTP client.""" self.client.close() def __enter__(self) -> "TradeMeClient": """Context manager entry.""" return self def __exit__(self, *args: Any) -> None: """Context manager exit.""" self.close()