Trade Me MCP Server

"""OAuth 1.0a authentication for Trade Me API.""" import os from typing import Optional from requests_oauthlib import OAuth1Session from oauthlib.oauth1 import Client import httpx class TradeMeAuth: """Handles OAuth 1.0a authentication for Trade Me API.""" def __init__( self, consumer_key: Optional[str] = None, consumer_secret: Optional[str] = None, access_token: Optional[str] = None, access_token_secret: Optional[str] = None, ): """ Initialize Trade Me OAuth authentication. Args: consumer_key: OAuth consumer key (defaults to TRADEME_CONSUMER_KEY env var) consumer_secret: OAuth consumer secret (defaults to TRADEME_CONSUMER_SECRET env var) access_token: OAuth access token (defaults to TRADEME_ACCESS_TOKEN env var) access_token_secret: OAuth access token secret (defaults to TRADEME_ACCESS_TOKEN_SECRET env var) """ self.consumer_key = consumer_key or os.getenv("TRADEME_CONSUMER_KEY") self.consumer_secret = consumer_secret or os.getenv("TRADEME_CONSUMER_SECRET") self.access_token = access_token or os.getenv("TRADEME_ACCESS_TOKEN") self.access_token_secret = access_token_secret or os.getenv( "TRADEME_ACCESS_TOKEN_SECRET" ) if not all( [self.consumer_key, self.consumer_secret] ): raise ValueError( "Consumer key and secret are required. " "Set TRADEME_CONSUMER_KEY and TRADEME_CONSUMER_SECRET environment variables." ) def get_auth_header(self, url: str, method: str = "GET") -> dict[str, str]: """ Generate OAuth 1.0a authentication headers for a request. Args: url: The URL to authenticate method: HTTP method (GET, POST, etc.) Returns: Dictionary containing the Authorization header """ client = Client( client_key=self.consumer_key, client_secret=self.consumer_secret, resource_owner_key=self.access_token, resource_owner_secret=self.access_token_secret, ) uri, headers, body = client.sign(url, http_method=method) return headers def is_authenticated(self) -> bool: """Check if we have access tokens for authenticated requests.""" return bool(self.access_token and self.access_token_secret) @property def auth(self) -> httpx.Auth: """ Get an httpx-compatible auth handler. Returns: Custom auth class for httpx """ return OAuth1Auth( self.consumer_key, self.consumer_secret, self.access_token, self.access_token_secret, ) class OAuth1Auth(httpx.Auth): """Custom OAuth1 auth handler for httpx.""" def __init__( self, consumer_key: str, consumer_secret: str, access_token: Optional[str] = None, access_token_secret: Optional[str] = None, ): self.consumer_key = consumer_key self.consumer_secret = consumer_secret self.access_token = access_token self.access_token_secret = access_token_secret def auth_flow(self, request: httpx.Request): """Apply OAuth1 authentication to the request.""" client = Client( client_key=self.consumer_key, client_secret=self.consumer_secret, resource_owner_key=self.access_token, resource_owner_secret=self.access_token_secret, ) url = str(request.url) method = request.method _, headers, _ = client.sign(url, http_method=method) # Update request headers with OAuth headers for key, value in headers.items(): request.headers[key] = value yield request