Use cases

export CARPE_API_KEY=cdm_your_key   # API → Keys
pip install openai requests          # Python recipes

import os, requests
from openai import OpenAI
 
BASE = "https://carpe-diem.xyz/api/operator"
 
# 1. Free quote: what would 500 tokens in / 300 out cost, on each market?
q = requests.post(f"{BASE}/router/quote", json={"model": "deepseek-v4-flash", "input_tokens": 500, "max_tokens": 300}).json()
print("best:", q["best"]["market"], q["best"]["costUsd"], "USD")
 
# 2. Run it through the router: the cheapest market serves it.
client = OpenAI(base_url=f"{BASE}/router", api_key=os.environ["CARPE_API_KEY"])
raw = client.chat.completions.with_raw_response.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Translate into German: The meeting moved to Thursday."}],
)
print(raw.parse().choices[0].message.content)
print("served by", raw.headers.get("x-carpe-route-market"), "for", raw.headers.get("x-carpe-route-cost-usdc"), "USDC")

import os, requests
from openai import OpenAI
 
BASE = "https://carpe-diem.xyz/api/operator"
models = requests.get(f"{BASE}/v1/models").json()["data"]
private = [m["id"] for m in models if m.get("carpe_diem_type") == "text" and m.get("privacy") == "private"]
print(len(private), "private text models, e.g.", private[:3])
 
client = OpenAI(base_url=f"{BASE}/v1", api_key=os.environ["CARPE_API_KEY"])
document = "This agreement is made between Alpha SA and Beta GmbH. Term: 24 months. Either party may terminate with 90 days' notice."
r = client.chat.completions.create(
    model="deepseek-v4-flash",                       # listed "private"
    messages=[
        {"role": "system", "content": "You summarise legal documents in five short bullet points."},
        {"role": "user", "content": document},
    ],
)
print(r.choices[0].message.content)

import base64, os, time, requests
 
BASE = "https://carpe-diem.xyz/api/operator/v1"
H = {"Authorization": f"Bearer {os.environ['CARPE_API_KEY']}"}
 
def queued(path, body):
    """Queue a slow image job, poll until the image comes back (binary)."""
    qid = requests.post(f"{BASE}/{path}/queue", headers=H, json=body).json()["queue_id"]
    while True:
        r = requests.post(f"{BASE}/{path}/retrieve", headers=H, json={"queue_id": qid})
        if r.headers.get("content-type", "").startswith("image/"):
            return r.content
        if r.json().get("status") != "pending":
            raise RuntimeError(r.text)
        time.sleep(2)
 
# 1. Generate (synchronous for fast models) → base64 WebP
gen = requests.post(f"{BASE}/image/generate", headers=H, json={
    "model": "z-image-turbo", "prompt": "a red fox in a forest, flat illustration",
}).json()
fox = gen["images"][0]
open("fox.webp", "wb").write(base64.b64decode(fox))
 
# 2. Edit (queued: edit models take longer) — the source goes in as a data URI
blue = queued("image/edit", {
    "model": "qwen-image-2-1-turbo-edit", "prompt": "make the fox blue",
    "image": f"data:image/webp;base64,{fox}",
})
open("fox-blue.png", "wb").write(blue)
 
# 3. Upscale ×2: raw base64, no data: prefix
up = requests.post(f"{BASE}/image/upscale", headers=H, json={
    "model": "upscaler", "image": base64.b64encode(blue).decode(), "scale": 2,
}, timeout=180)
up.raise_for_status()
open("fox-blue-2x.png", "wb").write(up.content)
print("cost of the last step:", up.headers.get("x-carpe-cost-usdc-micro"), "µUSDC")

import os, time, requests
 
BASE = "https://carpe-diem.xyz/api/operator"
H = {"Authorization": f"Bearer {os.environ['CARPE_API_KEY']}"}
job = {"model": "grok-imagine-1-5-lite-text-to-video", "prompt": "a paper boat on a calm lake, morning light",
       "duration": "5s", "aspect_ratio": "16:9"}
 
print("quote:", requests.post(f"{BASE}/v1/video/quote", headers=H, json=job).json()["cost_usdc"], "USD")
qid = requests.post(f"{BASE}/v1/video/queue", headers=H, json=job).json()["queue_id"]
 
while True:
    r = requests.post(f"{BASE}/v1/video/retrieve", headers=H, json={"queue_id": qid}).json()
    if r["status"] == "completed":
        url = r.get("video_url") or r.get("url")
        video = requests.get(url if url.startswith("http") else BASE + url)   # kept ~10 minutes
        open("boat.mp4", "wb").write(video.content)
        break
    if r["status"] == "failed":
        raise RuntimeError(r)
    time.sleep(5)

import os, requests
 
BASE = "https://carpe-diem.xyz/api/operator/v1"
H = {"Authorization": f"Bearer {os.environ['CARPE_API_KEY']}"}
 
speech = requests.post(f"{BASE}/audio/speech", headers=H, json={
    "model": "tts-kokoro", "input": "Bonjour ! Ceci est un test de synthèse vocale.", "response_format": "mp3",
})
speech.raise_for_status()
open("hello.mp3", "wb").write(speech.content)
 
text = requests.post(f"{BASE}/audio/transcriptions", headers=H,
    files={"file": ("hello.mp3", speech.content, "audio/mpeg")},
    data={"model": "openai/whisper-large-v3"}).json()
print(text["text"])

import os, time, requests
 
BASE = "https://carpe-diem.xyz/api/operator/v1"
H = {"Authorization": f"Bearer {os.environ['CARPE_API_KEY']}"}
 
qid = requests.post(f"{BASE}/audio/music/queue", headers=H, json={
    "model": "ace-step-15", "prompt": "calm lo-fi piano loop, warm, slow", "duration_seconds": 60,
}).json()["queue_id"]
 
while True:
    r = requests.post(f"{BASE}/audio/music/retrieve", headers=H, json={"queue_id": qid, "model": "ace-step-15"})
    if r.headers.get("content-type", "").startswith("audio/"):
        open("track.flac", "wb").write(r.content)  # ace-step-15 delivers FLAC; a /v1 track can be fetched once
        break
    if r.json().get("status") == "failed":
        raise RuntimeError(r.text)
    time.sleep(5)

import os, math
from openai import OpenAI
 
client = OpenAI(base_url="https://carpe-diem.xyz/api/operator/v1", api_key=os.environ["CARPE_API_KEY"])
docs = [
    "Refunds are possible within 30 days of purchase.",
    "Support answers in English and French, Monday to Friday.",
    "Enterprise plans include a dedicated account manager.",
]
 
def embed(texts):
    r = client.embeddings.create(model="text-embedding-bge-m3", input=texts)
    return [d.embedding for d in r.data]
 
def cos(a, b):
    return sum(x * y for x, y in zip(a, b)) / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)))
 
index = embed(docs)
question = "Can I get my money back after three weeks?"
q = embed([question])[0]
best = max(range(len(docs)), key=lambda i: cos(q, index[i]))
 
answer = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": f"Answer only from this context: {docs[best]}"},
        {"role": "user", "content": question},
    ],
)
print(answer.choices[0].message.content)

import os, requests
from openai import OpenAI
 
BASE = "https://carpe-diem.xyz/api/operator/v1"
H = {"Authorization": f"Bearer {os.environ['CARPE_API_KEY']}"}
 
results = requests.post(f"{BASE}/augment/search", headers=H, json={"query": "Base network latest upgrade"}).json()["results"][:5]
context = "\n\n".join(f"[{i + 1}] {r['title']} ({r['url']})\n{r['content']}" for i, r in enumerate(results))
 
client = OpenAI(base_url=BASE, api_key=os.environ["CARPE_API_KEY"])
r = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": "Answer from the sources below and cite them as [n].\n\n" + context},
        {"role": "user", "content": "What changed in the latest Base network upgrade?"},
    ],
)
print(r.choices[0].message.content)

import json, os, uuid
from openai import OpenAI
 
client = OpenAI(base_url="https://carpe-diem.xyz/api/operator/router", api_key=os.environ["CARPE_API_KEY"], max_retries=3)
schema = {
    "name": "invoice",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "vendor": {"type": "string"},
            "total": {"type": "number"},
            "currency": {"type": "string"},
            "due_date": {"type": "string"},
        },
        "required": ["vendor", "total", "currency", "due_date"],
        "additionalProperties": False,
    },
}
 
def extract(text: str) -> dict:
    r = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{"role": "user", "content": f"Extract the invoice fields:\n\n{text}"}],
        response_format={"type": "json_schema", "json_schema": schema},
        # Same key on a retry → the router replays the first answer instead of billing twice.
        extra_headers={"Idempotency-Key": str(uuid.uuid5(uuid.NAMESPACE_URL, text))},
    )
    return json.loads(r.choices[0].message.content)
 
print(extract("ACME Corp — Invoice 1042 — Total due: 1,250.00 EUR by 2026-12-01."))

curl -s https://carpe-diem.xyz/api/operator/v1/credits -H "Authorization: Bearer $CARPE_API_KEY"
curl -s "https://carpe-diem.xyz/api/operator/buyer/usage/summary?days=7" -H "Authorization: Bearer $CARPE_API_KEY"
curl -s "https://carpe-diem.xyz/api/operator/buyer/routing/summary" -H "Authorization: Bearer $CARPE_API_KEY"

{
  "mcpServers": {
    "carpe": {
      "command": "npx",
      "args": ["-y", "carpe-diem-mcp"],
      "env": {
        "CARPE_X402_PRIVATE_KEY": "0xYOUR_AGENT_WALLET_KEY",
        "CARPE_X402_BUDGET_USD": "2",
        "CARPE_MAX_PER_CALL_USD": "0.25"
      }
    }
  }
}