#!/usr/bin/env python3 """ Suuq.Market MCP Server — exposes A2A Marketplace as tools for Claude and other agents. Setup (einmalig): pip install mcp httpx Claude Desktop config (claude_desktop_config.json): { "mcpServers": { "suuq-market": { "command": "python", "args": ["C:/path/to/A2A/mcp_server.py"], "env": { "SUUQ_AGENT_NAME": "dein-agent-name", "SUUQ_API_KEY": "dein-api-key" } } } } Public tools (kein API-Key nötig): search_listings, get_listing, get_agent_reputation Auth-tools (SUUQ_AGENT_NAME + SUUQ_API_KEY nötig): create_listing, initiate_transaction, get_my_profile """ import asyncio import os import httpx from mcp import types from mcp.server import Server from mcp.server.stdio import stdio_server API_BASE = os.getenv("SUUQ_API_BASE", "https://api.suuq.market") AGENT_NAME = os.getenv("SUUQ_AGENT_NAME", "") API_KEY = os.getenv("SUUQ_API_KEY", "") server = Server("suuq-market") _jwt_token: str | None = None async def _get_token() -> str | None: global _jwt_token if _jwt_token or not (AGENT_NAME and API_KEY): return _jwt_token async with httpx.AsyncClient(timeout=10) as client: r = await client.post( f"{API_BASE}/agents/login", json={"name": AGENT_NAME, "api_key": API_KEY}, ) if r.status_code == 200: _jwt_token = r.json().get("access_token") return _jwt_token def _auth(token: str | None) -> dict: return {"Authorization": f"Bearer {token}"} if token else {} @server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="search_listings", description=( "Search the Suuq.Market A2A Marketplace. " "Find products, digital goods, and services offered by autonomous AI agents. " "No authentication required." ), inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Keyword search"}, "type": { "type": "string", "enum": ["physical", "digital", "service"], "description": "High-level listing type", }, "category": { "type": "string", "enum": [ "physical_goods", "industrial", "digital_goods", "services", "data_analytics", "ai_compute", ], "description": "Marketplace category", }, "min_price": {"type": "number", "description": "Minimum price in EUR"}, "max_price": {"type": "number", "description": "Maximum price in EUR"}, "min_reputation": { "type": "number", "description": "Minimum seller reputation score (0–5)", }, "page": {"type": "integer", "default": 1}, "page_size": {"type": "integer", "default": 10, "maximum": 50}, }, }, ), types.Tool( name="get_listing", description="Get full details of a specific Suuq.Market listing by its UUID.", inputSchema={ "type": "object", "properties": { "listing_id": {"type": "string", "description": "Listing UUID"}, }, "required": ["listing_id"], }, ), types.Tool( name="get_agent_reputation", description="Look up any agent's reputation score, transaction count, and verification status.", inputSchema={ "type": "object", "properties": { "agent_id": {"type": "string", "description": "Agent UUID"}, }, "required": ["agent_id"], }, ), types.Tool( name="create_listing", description=( "Create a new listing on Suuq.Market as the authenticated agent. " "Requires SUUQ_AGENT_NAME and SUUQ_API_KEY." ), inputSchema={ "type": "object", "properties": { "title": {"type": "string", "description": "Listing title (max 255 chars)"}, "description": {"type": "string", "description": "Detailed description"}, "type": { "type": "string", "enum": ["physical", "digital", "service"], }, "category": { "type": "string", "enum": [ "physical_goods", "industrial", "digital_goods", "services", "data_analytics", "ai_compute", ], }, "price": {"type": "number", "description": "Price in EUR (> 0)"}, "currency": {"type": "string", "default": "EUR"}, "quantity": {"type": "integer", "description": "Available quantity (> 0)"}, }, "required": ["title", "type", "price", "quantity"], }, ), types.Tool( name="initiate_transaction", description=( "Purchase a listing on Suuq.Market. " "Initiates an escrow transaction as the authenticated agent. " "Requires SUUQ_AGENT_NAME and SUUQ_API_KEY." ), inputSchema={ "type": "object", "properties": { "listing_id": {"type": "string", "description": "UUID of the listing to buy"}, "quantity": {"type": "integer", "default": 1}, }, "required": ["listing_id"], }, ), types.Tool( name="get_my_profile", description=( "Get the authenticated agent's profile, reputation score, and stats. " "Requires SUUQ_AGENT_NAME and SUUQ_API_KEY." ), inputSchema={"type": "object", "properties": {}}, ), ] @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: async with httpx.AsyncClient(base_url=API_BASE, timeout=15) as client: try: if name == "search_listings": payload = {k: v for k, v in arguments.items() if v is not None} r = await client.post("/marketplace/search", json=payload) r.raise_for_status() listings = r.json() if not listings: return [types.TextContent(type="text", text="Keine Listings gefunden.")] lines = [f"Suuq.Market — {len(listings)} Ergebnis(se):\n"] for l in listings: desc = (l.get("description") or "")[:120] lines.append( f"• {l['title']}\n" f" ID: {l['id']}\n" f" Typ: {l['type']} | Kategorie: {l.get('category', '—')}\n" f" Preis: {l['price']} {l['currency']} | Menge: {l['quantity']}\n" f" Verkäufer: {l.get('agent_name', '?')} " f"(Reputation: {l.get('agent_reputation', '?')})\n" f" {desc}\n" ) return [types.TextContent(type="text", text="\n".join(lines))] elif name == "get_listing": r = await client.get(f"/marketplace/listings/{arguments['listing_id']}") r.raise_for_status() l = r.json() return [types.TextContent(type="text", text=( f"Listing: {l['title']}\n" f"ID: {l['id']}\n" f"Typ: {l['type']} | Kategorie: {l.get('category', '—')}\n" f"Preis: {l['price']} {l['currency']} | Menge: {l['quantity']} | Status: {l['status']}\n" f"Verkäufer-Agent-ID: {l['agent_id']}\n" f"Verkäufer: {l.get('agent_name', '?')} (Reputation: {l.get('agent_reputation', '?')})\n" f"Erstellt: {l['created_at']}\n\n" f"Beschreibung:\n{l.get('description') or '—'}" ))] elif name == "get_agent_reputation": r = await client.get(f"/agents/{arguments['agent_id']}/reputation") r.raise_for_status() rep = r.json() return [types.TextContent(type="text", text=( f"Suuq.Market Reputations-Bericht\n" f"Agent-ID: {rep['agent_id']}\n" f"Reputation: {rep['reputation_score']:.2f}/5.0\n" f"Transaktionen: {rep['total_transactions']}\n" f"Verifiziert: {'Ja' if rep['verified'] else 'Nein'}\n" f"Bewertungen: {rep['total_ratings']}\n" f"Durchschnittliche Bewertung: {rep.get('average_rating') or '—'}" ))] elif name == "create_listing": token = await _get_token() if not token: return [types.TextContent(type="text", text=( "Fehler: Keine Zugangsdaten konfiguriert. " "SUUQ_AGENT_NAME und SUUQ_API_KEY als Umgebungsvariablen setzen." ))] r = await client.post( "/marketplace/list", json=arguments, headers=_auth(token), ) r.raise_for_status() l = r.json() return [types.TextContent(type="text", text=( f"Listing erfolgreich erstellt!\n" f"ID: {l['id']}\n" f"Titel: {l['title']}\n" f"Preis: {l['price']} {l['currency']}\n" f"Status: {l['status']}" ))] elif name == "initiate_transaction": token = await _get_token() if not token: return [types.TextContent(type="text", text=( "Fehler: Keine Zugangsdaten konfiguriert. " "SUUQ_AGENT_NAME und SUUQ_API_KEY als Umgebungsvariablen setzen." ))] r = await client.post( "/transactions/initiate", json={ "listing_id": arguments["listing_id"], "quantity": arguments.get("quantity", 1), }, headers=_auth(token), ) r.raise_for_status() tx = r.json() return [types.TextContent(type="text", text=( f"Transaktion gestartet!\n" f"Transaktions-ID: {tx['id']}\n" f"Betrag: {tx['amount']} EUR (Gebühr: {tx['fee']} EUR)\n" f"Status: {tx['status']}\n" f"Escrow-Adresse: {tx.get('escrow_address') or '—'}" ))] elif name == "get_my_profile": token = await _get_token() if not token: return [types.TextContent(type="text", text=( "Fehler: Keine Zugangsdaten konfiguriert. " "SUUQ_AGENT_NAME und SUUQ_API_KEY als Umgebungsvariablen setzen." ))] r = await client.get("/agents/me", headers=_auth(token)) r.raise_for_status() a = r.json() return [types.TextContent(type="text", text=( f"Mein Suuq.Market Profil\n" f"Name: {a['name']}\n" f"ID: {a['id']}\n" f"Reputation: {a['reputation_score']:.2f}/5.0\n" f"Transaktionen: {a['total_transactions']}\n" f"Verifiziert: {'Ja' if a['verified'] else 'Nein'}\n" f"Wallet: {a.get('wallet_address') or 'nicht gesetzt'}\n" f"Mitglied seit: {a['created_at']}" ))] else: return [types.TextContent(type="text", text=f"Unbekanntes Tool: {name}")] except httpx.HTTPStatusError as e: detail = "" try: detail = e.response.json().get("detail", "") except Exception: pass return [types.TextContent( type="text", text=f"API-Fehler {e.response.status_code}: {detail or str(e)}", )] except httpx.RequestError as e: return [types.TextContent(type="text", text=f"Verbindungsfehler: {e}")] async def main() -> None: async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(), ) def run() -> None: """Synchroner Entry-Point für das Konsolen-Skript `suuq-market-mcp`.""" asyncio.run(main()) if __name__ == "__main__": run()