#!/usr/bin/env python3 """ eBay Browse API MCP Server & CLI Tool Connects to eBay's official Browse API using OAuth 2.0 Client Credentials Grant. Allows AI assistants to search items on eBay.de, filter by condition (NEW, USED, REFURBISHED), buying format (Sofort-Kaufen / Auktion), price range, and fetch listing details. Runs standalone over MCP stdio or direct CLI commands. Requires zero external dependencies (pure Python standard library). """ import os import sys import json import time import base64 import urllib.request import urllib.parse from typing import Dict, Any, List, Optional EBAY_TOKEN_URL = "https://api.ebay.com/identity/v1/oauth2/token" EBAY_BROWSE_URL = "https://api.ebay.com/buy/browse/v1" EBAY_OAUTH_SCOPE = "https://api.ebay.com/oauth/api_scope" EBAY_APP_ID = os.environ.get("EBAY_APP_ID", "").strip() EBAY_CERT_ID = os.environ.get("EBAY_CERT_ID", "").strip() EBAY_MARKETPLACE_ID = os.environ.get("EBAY_MARKETPLACE_ID", "EBAY_DE") # Condition mapping for user-friendly terms CONDITION_MAP = { "new": "NEW", "neu": "NEW", "used": "USED", "gebraucht": "USED", "refurbished": "CERTIFIED_REFURBISHED,EXCELLENT_REFURBISHED,VERY_GOOD_REFURBISHED,GOOD_REFURBISHED,SELLER_REFURBISHED", "generalüberholt": "CERTIFIED_REFURBISHED,EXCELLENT_REFURBISHED,VERY_GOOD_REFURBISHED,GOOD_REFURBISHED,SELLER_REFURBISHED", } BUYING_FORMAT_MAP = { "fixed_price": "FIXED_PRICE", "sofortkauf": "FIXED_PRICE", "auction": "AUCTION", "auktion": "AUCTION", } class EbayClient: def __init__(self, app_id: Optional[str] = None, cert_id: Optional[str] = None, marketplace_id: Optional[str] = None): self.app_id = app_id or EBAY_APP_ID self.cert_id = cert_id or EBAY_CERT_ID self.marketplace_id = marketplace_id or EBAY_MARKETPLACE_ID self._access_token: Optional[str] = None self._token_expires_at: float = 0.0 def is_configured(self) -> bool: dummy_values = ["", "dummy-ebay-app-id", "dummy-ebay-cert-id", "your-ebay-app-id", "your-ebay-cert-id"] return bool(self.app_id and self.cert_id and self.app_id not in dummy_values and self.cert_id not in dummy_values) def get_token(self) -> str: """Retrieves or refreshes eBay OAuth 2.0 application access token.""" if not self.is_configured(): raise ValueError( "eBay Developer credentials (EBAY_APP_ID and EBAY_CERT_ID) are not configured. " "Please add them to openclaw.secret.yaml under ebay-app-id and ebay-cert-id." ) # Reuse cached token if valid (with 60-second buffer) if self._access_token and time.time() < self._token_expires_at - 60: return self._access_token creds = f"{self.app_id}:{self.cert_id}".encode("utf-8") basic_auth = base64.b64encode(creds).decode("utf-8") data = urllib.parse.urlencode({ "grant_type": "client_credentials", "scope": EBAY_OAUTH_SCOPE }).encode("utf-8") req = urllib.request.Request( EBAY_TOKEN_URL, data=data, headers={ "Authorization": f"Basic {basic_auth}", "Content-Type": "application/x-www-form-urlencoded" } ) with urllib.request.urlopen(req, timeout=15) as resp: body = json.loads(resp.read().decode("utf-8")) self._access_token = body.get("access_token") expires_in = body.get("expires_in", 7200) self._token_expires_at = time.time() + expires_in return self._access_token def search( self, query: str, condition: Optional[str] = None, buying_format: Optional[str] = None, min_price: Optional[float] = None, max_price: Optional[float] = None, limit: int = 15 ) -> Dict[str, Any]: """Searches items on eBay via the Browse API.""" if not self.is_configured(): return { "configured": False, "error": "eBay credentials not configured. Please set EBAY_APP_ID and EBAY_CERT_ID in openclaw.secret.yaml.", "query": query, "results": [] } try: token = self.get_token() except Exception as e: return { "configured": False, "error": f"eBay authentication failed: {str(e)}", "query": query, "results": [] } # Build query filters filters = [] if min_price is not None or max_price is not None: min_str = f"{min_price:.2f}" if min_price is not None else "" max_str = f"{max_price:.2f}" if max_price is not None else "" filters.append(f"price:[{min_str}..{max_str}]") if condition: cond_lower = condition.lower().strip() ebay_cond = CONDITION_MAP.get(cond_lower, condition.upper()) filters.append(f"conditions:{{{ebay_cond}}}") if buying_format: fmt_lower = buying_format.lower().strip() ebay_fmt = BUYING_FORMAT_MAP.get(fmt_lower, buying_format.upper()) filters.append(f"buyingOptions:{{{ebay_fmt}}}") params = { "q": query, "limit": str(min(max(limit, 1), 50)) } if filters: params["filter"] = ",".join(filters) url = f"{EBAY_BROWSE_URL}/item_summary/search?{urllib.parse.urlencode(params)}" req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {token}", "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id, "Accept": "application/json" } ) try: with urllib.request.urlopen(req, timeout=15) as resp: data = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: err_text = e.read().decode("utf-8", errors="replace") return {"error": f"eBay API HTTP error {e.code}: {err_text}", "query": query, "results": []} except Exception as e: return {"error": f"eBay search error: {str(e)}", "query": query, "results": []} items = data.get("itemSummaries", []) total = data.get("total", len(items)) results = [] for it in items: price_dict = it.get("price") or {} price_val = price_dict.get("value") price_curr = price_dict.get("currency", "EUR") price_str = f"{float(price_val):.2f} {price_curr}" if price_val is not None else "Kein Preis" if price_curr == "EUR" and price_val is not None: price_str = f"{float(price_val):.2f} €".replace(".", ",") # Shipping options shipping_list = it.get("shippingOptions") or [] shipping_info = "Versand nicht angegeben" shipping_cost = None if shipping_list: s_cost_dict = shipping_list[0].get("shippingCost") or {} s_val = s_cost_dict.get("value") if s_val == "0.00" or s_val == "0" or s_val is None: shipping_info = "Kostenloser Versand" shipping_cost = 0.0 else: shipping_cost = float(s_val) shipping_info = f"{shipping_cost:.2f} € Versand".replace(".", ",") # Buying options (Auktion vs Sofort-Kaufen) buying_opts = it.get("buyingOptions") or [] is_auction = "AUCTION" in buying_opts is_fixed_price = "FIXED_PRICE" in buying_opts format_label = "Sofort-Kaufen" if is_fixed_price else ("Auktion" if is_auction else ", ".join(buying_opts)) seller = it.get("seller") or {} results.append({ "item_id": it.get("itemId"), "title": it.get("title"), "price": price_str, "price_numeric": float(price_val) if price_val is not None else None, "shipping": shipping_info, "shipping_cost": shipping_cost, "condition": it.get("condition"), "format": format_label, "seller_username": seller.get("username"), "seller_feedback_percent": seller.get("feedbackPercentage"), "url": it.get("itemWebUrl"), "image_url": it.get("image", {}).get("imageUrl") }) return { "query": query, "marketplace": self.marketplace_id, "total_found": total, "returned_count": len(results), "results": results } def get_item(self, item_id: str) -> Dict[str, Any]: """Fetches full details for a specific eBay listing.""" if not self.is_configured(): return {"configured": False, "error": "eBay credentials not configured."} try: token = self.get_token() except Exception as e: return {"configured": False, "error": f"eBay authentication failed: {str(e)}"} url = f"{EBAY_BROWSE_URL}/item/{urllib.parse.quote(item_id)}" req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {token}", "X-EBAY-C-MARKETPLACE-ID": self.marketplace_id, "Accept": "application/json" } ) try: with urllib.request.urlopen(req, timeout=15) as resp: item = json.loads(resp.read().decode("utf-8")) except Exception as e: return {"error": f"Failed to fetch eBay item {item_id}: {str(e)}"} images = [] if item.get("image", {}).get("imageUrl"): images.append(item["image"]["imageUrl"]) for add_img in item.get("additionalImages", []): if add_img.get("imageUrl"): images.append(add_img["imageUrl"]) price_dict = item.get("price") or {} price_val = price_dict.get("value") price_curr = price_dict.get("currency", "EUR") price_str = f"{float(price_val):.2f} €".replace(".", ",") if price_val else "Kein Preis" seller = item.get("seller") or {} location = item.get("itemLocation") or {} loc_str = ", ".join(filter(None, [location.get("city"), location.get("stateOrProvince"), location.get("country")])) return { "item_id": item.get("itemId"), "title": item.get("title"), "price": price_str, "condition": item.get("condition"), "description": item.get("shortDescription") or (item.get("description", "")[:400] if item.get("description") else None), "location": loc_str or None, "seller": seller.get("username"), "seller_feedback_score": seller.get("feedbackScore"), "seller_feedback_percent": seller.get("feedbackPercentage"), "images": images, "url": item.get("itemWebUrl") } # ============================================================================== # Model Context Protocol (MCP) stdio Server # ============================================================================== MCP_TOOLS = [ { "name": "ebay_search", "description": "Search items on eBay.de via official Browse API. Supports filtering by condition (NEW, USED, REFURBISHED), format (AUCTION, FIXED_PRICE/Sofort-Kaufen), and price bounds.", "inputSchema": { "type": "object", "properties": { "query": { "type": "string", "description": "Item query, e.g. 'ThinkPad T14', 'iPhone 13 gebraucht', 'Sonos One'" }, "condition": { "type": "string", "description": "Item condition: 'used'/'gebraucht', 'refurbished'/'generalüberholt', or 'new'/'neu'", "enum": ["used", "gebraucht", "refurbished", "generalüberholt", "new", "neu"] }, "buying_format": { "type": "string", "description": "Buying format: 'fixed_price'/'sofortkauf' or 'auction'/'auktion'", "enum": ["fixed_price", "sofortkauf", "auction", "auktion"] }, "min_price": { "type": "number", "description": "Minimum price in EUR" }, "max_price": { "type": "number", "description": "Maximum price in EUR" }, "limit": { "type": "integer", "description": "Max number of items to return (default 10)", "default": 10 } }, "required": ["query"] } }, { "name": "ebay_get_item", "description": "Fetch detailed information for a specific eBay listing (description, item location, additional photos, return policy).", "inputSchema": { "type": "object", "properties": { "item_id": { "type": "string", "description": "eBay listing item ID, obtained from ebay_search" } }, "required": ["item_id"] } } ] def run_mcp_server(): client = EbayClient() while True: line = sys.stdin.readline() if not line: break line = line.strip() if not line: continue try: req = json.loads(line) except Exception as e: sys.stderr.write(f"Invalid JSON received: {e}\n") continue msg_id = req.get("id") method = req.get("method") params = req.get("params", {}) if method == "initialize": resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "ebay-mcp", "version": "1.0.0"} } } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() elif method == "notifications/initialized": pass elif method == "ping": resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}} sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() elif method == "tools/list": resp = { "jsonrpc": "2.0", "id": msg_id, "result": {"tools": MCP_TOOLS} } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() elif method == "tools/call": tool_name = params.get("name") args = params.get("arguments", {}) try: if tool_name == "ebay_search": data = client.search( query=args.get("query", ""), condition=args.get("condition"), buying_format=args.get("buying_format"), min_price=args.get("min_price"), max_price=args.get("max_price"), limit=args.get("limit", 10) ) elif tool_name == "ebay_get_item": data = client.get_item(item_id=args.get("item_id", "")) else: raise ValueError(f"Unknown tool: {tool_name}") text_result = json.dumps(data, indent=2, ensure_ascii=False) resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "content": [{"type": "text", "text": text_result}] } } except Exception as e: resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "isError": True, "content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}] } } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() else: if msg_id is not None: resp = { "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"} } sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() def run_cli(): import argparse parser = argparse.ArgumentParser(description="eBay Browse API MCP & CLI") subparsers = parser.add_subparsers(dest="command") sp_search = subparsers.add_parser("search", help="Search items on eBay") sp_search.add_argument("query", help="Item search query") sp_search.add_argument("--condition", choices=["used", "gebraucht", "refurbished", "new", "neu"]) sp_search.add_argument("--format", choices=["fixed_price", "sofortkauf", "auction", "auktion"]) sp_search.add_argument("--min-price", type=float) sp_search.add_argument("--max-price", type=float) sp_search.add_argument("--limit", type=int, default=10) sp_item = subparsers.add_parser("item", help="Get eBay item details") sp_item.add_argument("item_id", help="eBay Item ID") args = parser.parse_args() client = EbayClient() if args.command == "search": print(json.dumps(client.search( query=args.query, condition=args.condition, buying_format=args.format, min_price=args.min_price, max_price=args.max_price, limit=args.limit ), indent=2, ensure_ascii=False)) elif args.command == "item": print(json.dumps(client.get_item(args.item_id), indent=2, ensure_ascii=False)) else: parser.print_help() if __name__ == "__main__": if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"): run_mcp_server() else: run_cli()