leora/agent-market
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
agent-market/server.py

377 lines
22 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Agent Market demo server. Tiny on purpose: static files, an LLM proxy (key stays server-side),
test-mode payment endpoints, and a small JSON action API shared by the chat agent and outside agents."""
import json, os, re, secrets, sys, threading, time, urllib.parse
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from market.store import Store, now, new_id
from market import agent as A
from market import payments
HOST = os.environ.get("AGENT_MARKET_HOST", "127.0.0.1")
PORT = int(os.environ.get("AGENT_MARKET_PORT", "8796"))
DATA_DIR = os.environ.get("AGENT_MARKET_DATA", os.path.join(os.path.dirname(os.path.abspath(__file__)), "data"))
ENV_LABEL = os.environ.get("AGENT_MARKET_ENV", "demo")
STATIC = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static")
WEBHOOK_SECRET = os.environ.get("AGENT_MARKET_WEBHOOK_SECRET") or secrets.token_hex(16) # per-process if unset
MAX_BODY = 64 * 1024
CHAT_RPM = int(os.environ.get("AGENT_MARKET_CHAT_RPM", "12")) # per-IP chat requests per minute
CHAT_BURST = int(os.environ.get("AGENT_MARKET_CHAT_BURST", "6"))
_buckets = {} # ip -> [tokens, last_ts]
_bucket_lock = threading.Lock()
def rate_ok(ip):
"""Token bucket per client IP. Returns (allowed, retry_after_seconds)."""
if CHAT_RPM <= 0:
return True, 0
rate = CHAT_RPM / 60.0
with _bucket_lock:
now_t = time.time()
tokens, last = _buckets.get(ip, (float(CHAT_BURST), now_t))
tokens = min(float(CHAT_BURST), tokens + (now_t - last) * rate)
if tokens >= 1:
_buckets[ip] = (tokens - 1, now_t)
return True, 0
_buckets[ip] = (tokens, now_t)
if len(_buckets) > 5000:
_buckets.clear()
return False, int((1 - tokens) / rate) + 1
store = Store(DATA_DIR)
ACTIONS = {
"search": "Search active listings. args: {q?, kind?: product|service|any, limit?}",
"get_listing": "Get one listing. args: {id}",
"create_listing": "Sell something. args: {title, kind: product|service, price (cents), description, category?, area?, payment_options?: [online,cash], emoji?}",
"quote": "Price breakdown online vs cash. args: {listing_id, quantity?}",
"start_checkout": "Start an online (TEST MODE) purchase; returns a payment intent with the fee. args: {listing_id, quantity?}",
"confirm_payment": "Pay a payment intent with a TEST card. args: {intent_id, card_number: '4242424242424242'}",
"start_cash_deal": "Arrange a cash-in-person deal (no fee). args: {listing_id, place?, time?}",
"complete_cash_deal": "Mark a cash deal as paid in person (fee stays $0). args: {deal_id}",
"list_orders": "Orders where you are buyer or seller. args: {}",
}
def quote(listing_id, qty=1):
L = store.get("listings", listing_id)
if not L:
return None
sub = L["price"] * max(1, int(qty or 1))
fee = A.fee_for(sub)
return {"listing_id": L["id"], "title": L["title"], "currency": "USD", "quantity": max(1, int(qty or 1)),
"online": {"available": "online" in L.get("payment_options", []), "subtotal_cents": sub, "fee_cents": fee,
"fee_bps": A.FEE_BPS, "total_cents": sub + fee, "seller_receives_cents": sub, "mode": "test"},
"cash_in_person": {"available": "cash" in L.get("payment_options", []), "subtotal_cents": sub,
"fee_cents": 0, "total_cents": sub, "seller_receives_cents": sub}}
def confirm_payment(intent_id, card_number):
intent = store.get("intents", intent_id)
if not intent:
return 404, {"error": "not_found"}
if intent["status"] == "succeeded":
return 200, {"status": "succeeded", "intent": intent, "order": store.get("orders", intent.get("order_id"))}
res = payments.charge(intent, card_number, f"http://127.0.0.1:{PORT}/api/payments/webhook", WEBHOOK_SECRET)
intent = store.get("intents", intent_id)
return 200, {"status": res["status"], "error": res.get("error"), "message": res.get("message"),
"intent": intent, "order": store.get("orders", intent.get("order_id")) if intent.get("order_id") else None,
"test_mode": True}
def run_action(action, args, actor):
args = dict(args or {})
if action == "search":
return 200, {"listings": A.run_tools(store, [{"name": "search", "args": args}])[0]["listings"]}
if action == "get_listing":
L = store.get("listings", args.get("id"))
return (200, {"listing": L}) if L else (404, {"error": "not_found"})
if action == "create_listing":
args.update(seller=actor, seller_type="agent" if actor.startswith("agent:") else "human")
r = A.run_tools(store, [{"name": "create_listing", "args": args}])[0]
return (201, {"listing": r["listing"]}) if r["ok"] else (400, {"error": r["error"]})
if action == "quote":
q = quote(args.get("listing_id"), args.get("quantity"))
return (200, q) if q else (404, {"error": "not_found"})
if action == "start_checkout":
args["buyer"] = actor
r = A.run_tools(store, [{"name": "start_checkout", "args": args}])[0]
return (201, {"intent": r["intent"], "next": {"action": "confirm_payment", "args": {"intent_id": r["intent"]["id"], "card_number": "4242424242424242"}}}) if r["ok"] else (400, {"error": r["error"]})
if action == "confirm_payment":
return confirm_payment(args.get("intent_id"), args.get("card_number"))
if action == "start_cash_deal":
args["buyer"] = actor
r = A.run_tools(store, [{"name": "start_cash_deal", "args": args}])[0]
return (201, {"deal": r["deal"]}) if r["ok"] else (400, {"error": r["error"]})
if action == "complete_cash_deal":
r = A.run_tools(store, [{"name": "complete_cash_deal", "args": args}])[0]
return (200, {"deal": r["deal"], "order": r["order"]}) if r["ok"] else (404, {"error": r["error"]})
if action == "list_orders":
return 200, {"orders": [o for o in store.all("orders") if actor in (o.get("buyer"), o.get("seller"))]}
return 400, {"error": "unknown_action", "actions": list(ACTIONS)}
def hydrate(actions, actor):
out = []
for a in actions or []:
if not isinstance(a, dict) or a.get("type") not in A.ALLOWED_ACTIONS:
continue
p = dict(a.get("payload") or {})
t = a["type"]
if t in ("show_cards", "show_compare"):
ids = [i for i in (p.get("listing_ids") or []) if isinstance(i, str)][:12]
p["listings"] = [x for x in (store.get("listings", i) for i in ids) if x]
if t == "show_compare":
p["quotes"] = [quote(x["id"]) for x in p["listings"]]
if not p["listings"]:
continue
elif t == "show_checkout":
intent = store.get("intents", p.get("intent_id") or "")
if not intent:
r = A.run_tools(store, [{"name": "start_checkout", "args": {"listing_id": p.get("listing_id"), "buyer": actor, "quantity": p.get("quantity") or 1}}])[0]
if not r["ok"]:
out.append({"type": "toast", "payload": {"level": "warn", "text": f"Checkout unavailable: {r['error']}"}})
continue
intent = r["intent"]
p.update(intent=intent, listing=store.get("listings", intent["listing_id"]), intent_id=intent["id"])
elif t == "show_cash_meetup":
deal = store.get("cash_deals", p.get("deal_id") or "")
if not deal:
r = A.run_tools(store, [{"name": "start_cash_deal", "args": {"listing_id": p.get("listing_id"), "buyer": actor, "place": p.get("suggested_place"), "time": p.get("suggested_time")}}])[0]
if not r["ok"]:
out.append({"type": "toast", "payload": {"level": "warn", "text": f"Cash deal unavailable: {r['error']}"}})
continue
deal = r["deal"]
p.update(deal=deal, listing=store.get("listings", deal["listing_id"]), deal_id=deal["id"])
elif t == "show_receipt":
o = store.get("orders", p.get("order_id") or "")
if not o:
continue
p["order"] = o
out.append({"type": t, "payload": p})
return out
def agent_manifest(base):
return {
"schema": "agent-market/agent.json v1",
"name": "Agent Market (demo)",
"description": "Demo marketplace where people and AI agents list, buy and sell products and services. TEST MODE ONLY: no real money moves.",
"test_mode": True,
"base_url": base,
"fees": {"online_payment_fee_bps": A.FEE_BPS, "online_payment_fee_percent": A.FEE_BPS / 100,
"cash_in_person_fee_bps": 0, "who_pays": "buyer pays the facilitation fee on top of the price; seller receives the full price"},
"payments": {"provider": "mock (test mode)", "test_cards": {"success": "4242424242424242", "declined": "4000000000000002", "insufficient_funds": "4000000000009995"}},
"identity": "Send header X-Agent-Id: agent:<your-name>. Demo only: identity is not verified.",
"endpoint": {"method": "POST", "url": base + "/api/actions", "body": {"action": "<name>", "args": {}}},
"actions": ACTIONS,
"rest_aliases": ["GET /api/listings?q=", "GET /api/listings/{id}", "POST /api/listings", "GET /api/quote/{id}",
"POST /api/checkout", "POST /api/payments/{intent_id}/confirm", "POST /api/cash-deals",
"POST /api/cash-deals/{id}/complete", "GET /api/orders"],
"openapi": base + "/openapi.json",
"example_flow": [
{"action": "search", "args": {"q": "bike"}},
{"action": "start_checkout", "args": {"listing_id": "lst_..."}},
{"action": "confirm_payment", "args": {"intent_id": "pi_...", "card_number": "4242424242424242"}},
],
}
def openapi(base):
act_enum = list(ACTIONS)
return {
"openapi": "3.0.3",
"info": {"title": "Agent Market demo API", "version": "0.1.0",
"description": "TEST MODE ONLY. Same actions the chat agent uses. Online fee %.2f%%, cash in person 0%%." % (A.FEE_BPS / 100)},
"servers": [{"url": base}],
"paths": {
"/api/actions": {"post": {"summary": "Run a marketplace action", "parameters": [{"name": "X-Agent-Id", "in": "header", "schema": {"type": "string"}, "description": "agent:<name>"}],
"requestBody": {"required": True, "content": {"application/json": {"schema": {"type": "object", "required": ["action"], "properties": {"action": {"type": "string", "enum": act_enum}, "args": {"type": "object"}}}}}},
"responses": {"200": {"description": "Result"}, "201": {"description": "Created"}, "400": {"description": "Bad request"}, "404": {"description": "Not found"}}}},
"/api/listings": {"get": {"summary": "Search listings", "parameters": [{"name": "q", "in": "query", "schema": {"type": "string"}}, {"name": "kind", "in": "query", "schema": {"type": "string", "enum": ["product", "service", "any"]}}], "responses": {"200": {"description": "Listings"}}},
"post": {"summary": "Create a listing", "responses": {"201": {"description": "Created"}}}},
"/api/listings/{id}": {"get": {"summary": "Get a listing", "parameters": [{"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "Listing"}}}},
"/api/quote/{id}": {"get": {"summary": "Online vs cash price breakdown", "parameters": [{"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "Quote"}}}},
"/api/checkout": {"post": {"summary": "Create a test-mode payment intent (adds fee)", "responses": {"201": {"description": "Intent"}}}},
"/api/payments/{intent_id}/confirm": {"post": {"summary": "Pay with a TEST card", "parameters": [{"name": "intent_id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "Payment result"}}}},
"/api/cash-deals": {"post": {"summary": "Arrange cash in person (no fee)", "responses": {"201": {"description": "Deal"}}}},
"/api/cash-deals/{id}/complete": {"post": {"summary": "Mark cash deal completed", "parameters": [{"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}], "responses": {"200": {"description": "Order"}}}},
"/api/orders": {"get": {"summary": "Your orders", "responses": {"200": {"description": "Orders"}}}},
"/api/chat": {"post": {"summary": "Talk to the UI agent (returns reply + UI actions)", "responses": {"200": {"description": "Agent reply"}}}},
},
}
class H(BaseHTTPRequestHandler):
server_version = "AgentMarket/0.1"
def log_message(self, fmt, *args):
sys.stderr.write("[%s] %s\n" % (now(), fmt % args))
def _base(self):
proto = self.headers.get("X-Forwarded-Proto") or "http"
host = self.headers.get("X-Forwarded-Host") or self.headers.get("Host") or f"127.0.0.1:{PORT}"
return f"{proto}://{host}" if "trycloudflare" not in host else f"https://{host}"
def _actor(self, body=None):
a = self.headers.get("X-Agent-Id") or (body or {}).get("actor") or "guest"
return re.sub(r"[^A-Za-z0-9:_\-. ()]", "", str(a))[:60] or "guest"
def _client_ip(self):
# Only trust proxy headers when the peer is the local tunnel (cloudflared) on loopback.
peer = self.client_address[0]
if peer in ("127.0.0.1", "::1"):
fwd = self.headers.get("CF-Connecting-IP") or (self.headers.get("X-Forwarded-For") or "").split(",")[0].strip()
if fwd:
return fwd
return peer
def _json(self, code, obj, extra_headers=None):
b = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(b)))
self.send_header("Cache-Control", "no-store")
self.send_header("Access-Control-Allow-Origin", "*")
for k, v in (extra_headers or {}).items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(b)
def _body(self):
n = int(self.headers.get("Content-Length") or 0)
if n > MAX_BODY:
return None, b""
raw = self.rfile.read(n) if n else b""
try:
return (json.loads(raw) if raw else {}), raw
except Exception:
return None, raw
def do_OPTIONS(self):
self.send_response(204)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Agent-Id")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.end_headers()
def _static(self, rel):
path = os.path.normpath(os.path.join(STATIC, rel))
if not path.startswith(STATIC) or not os.path.isfile(path):
return self._json(404, {"error": "not_found"})
ctype = {".html": "text/html; charset=utf-8", ".js": "application/javascript", ".css": "text/css",
".svg": "image/svg+xml", ".png": "image/png"}.get(os.path.splitext(path)[1], "application/octet-stream")
with open(path, "rb") as f:
b = f.read()
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(b)))
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(b)
def do_GET(self):
u = urllib.parse.urlparse(self.path)
path, qs = u.path, dict(urllib.parse.parse_qsl(u.query))
if path in ("/", "/index.html"):
return self._static("index.html")
if path.startswith("/static/"):
return self._static(path[len("/static/"):])
if path in ("/agent.json", "/.well-known/agent.json"):
return self._json(200, agent_manifest(self._base()))
if path == "/openapi.json":
return self._json(200, openapi(self._base()))
if path == "/healthz":
return self._json(200, {"ok": True})
if path == "/api/status":
return self._json(200, {"env": ENV_LABEL, "agent": A.agent_status(), "fee_bps": A.FEE_BPS,
"payments": {"mode": "test", "provider": "mock", "test_card": "4242 4242 4242 4242"},
"counts": {k: len(store.all(k)) for k in ("listings", "intents", "orders", "cash_deals")}})
if path == "/api/listings":
return self._json(*run_action("search", {"q": qs.get("q", ""), "kind": qs.get("kind", "any"), "limit": int(qs.get("limit", 50))}, self._actor()))
m = re.fullmatch(r"/api/listings/([a-z0-9_]+)", path)
if m:
return self._json(*run_action("get_listing", {"id": m.group(1)}, self._actor()))
m = re.fullmatch(r"/api/quote/([a-z0-9_]+)", path)
if m:
return self._json(*run_action("quote", {"listing_id": m.group(1), "quantity": qs.get("quantity", 1)}, self._actor()))
if path == "/api/orders":
return self._json(*run_action("list_orders", {}, self._actor()))
return self._json(404, {"error": "not_found"})
def do_POST(self):
path = urllib.parse.urlparse(self.path).path
body, raw = self._body()
if body is None:
return self._json(400, {"error": "bad_json_or_too_large"})
if path == "/api/payments/webhook":
sig = self.headers.get("X-Mock-Signature") or ""
if not payments.hmac.compare_digest(sig, payments.sign(WEBHOOK_SECRET, raw)):
return self._json(401, {"error": "bad_signature"})
return self._json(200, self._webhook(body))
actor = self._actor(body)
if path == "/api/chat":
# Only free-text turns reach the LLM; deterministic UI events (clicks, payments) are not limited.
llm_bound = not isinstance(body.get("event"), dict) or A.LLM_EVENTS
ok, retry = rate_ok(self._client_ip()) if llm_bound else (True, 0)
if not ok:
return self._json(429, {"error": "rate_limited", "retry_after": retry, "agent": "rate-limit",
"reply": f"You're sending messages quickly. Please wait about {retry}s and try again.",
"actions": []}, {"Retry-After": str(retry)})
msg = str(body.get("message") or "")[:2000]
ev = body.get("event") if isinstance(body.get("event"), dict) else None
hist = body.get("history") if isinstance(body.get("history"), list) else []
out = A.handle_chat(store, msg, hist, ev, actor)
out["actions"] = hydrate(out.get("actions"), actor)
out.pop("tools", None)
return self._json(200, out)
if path == "/api/actions":
return self._json(*run_action(str(body.get("action")), body.get("args") or {}, actor))
if path == "/api/listings":
return self._json(*run_action("create_listing", body, actor))
if path == "/api/checkout":
return self._json(*run_action("start_checkout", body, actor))
m = re.fullmatch(r"/api/payments/([a-z0-9_]+)/confirm", path)
if m:
return self._json(*confirm_payment(m.group(1), body.get("card_number")))
if path == "/api/cash-deals":
return self._json(*run_action("start_cash_deal", body, actor))
m = re.fullmatch(r"/api/cash-deals/([a-z0-9_]+)/complete", path)
if m:
return self._json(*run_action("complete_cash_deal", {"deal_id": m.group(1)}, actor))
return self._json(404, {"error": "not_found"})
def _webhook(self, ev):
d = ev.get("data") or {}
intent = store.get("intents", d.get("intent_id") or "")
if not intent:
return {"received": True, "ignored": "unknown_intent"}
if ev.get("type") == "payment_intent.succeeded" and intent["status"] != "succeeded":
L = store.get("listings", intent["listing_id"])
order = {"id": new_id("ord"), "listing_id": intent["listing_id"], "title": intent["title"],
"buyer": intent["buyer"], "seller": (L or {}).get("seller"), "method": "online_test_card",
"subtotal_cents": intent["subtotal_cents"], "fee_cents": intent["fee_cents"], "fee_bps": intent["fee_bps"],
"total_cents": intent["total_cents"], "seller_receives_cents": intent["subtotal_cents"],
"card_last4": d.get("card_last4"), "currency": "USD", "status": "paid", "mode": "test",
"provider_event": ev.get("id"), "created_at": now()}
store.put("orders", order)
intent.update(status="succeeded", order_id=order["id"], paid_at=now())
store.put("intents", intent)
if L and L.get("kind") == "product":
L["status"] = "sold"
store.put("listings", L)
store.log("payment_succeeded", {"intent": intent["id"], "order": order["id"]})
elif ev.get("type") == "payment_intent.payment_failed":
intent.update(status="requires_payment", last_error=d.get("failure_code"))
store.put("intents", intent)
store.log("payment_failed", {"intent": intent["id"], "code": d.get("failure_code")})
return {"received": True}
if __name__ == "__main__":
srv = ThreadingHTTPServer((HOST, PORT), H)
st = A.agent_status()
print(f"Agent Market [{ENV_LABEL}] on http://{HOST}:{PORT} agent={st['mode']} fee={A.FEE_BPS/100:.2f}% payments=TEST(mock)", flush=True)
srv.serve_forever()
Reference in New Issue View Git Blame Copy Permalink