Langfuse tells you what your agents did and how well they did it. Purple Flea pays them for it. Close the loop from LLM observability to financial settlement without writing custom billing infrastructure.
Langfuse captures every LLM call, tool use, and user interaction as a structured trace. Purple Flea reads that trace — either via webhook or direct API call — and uses the data to create, condition, and release escrow payments automatically.
The Langfuse LiteLLM callback handler captures every LLM call automatically. Attach a Purple Flea escrow creation to the same workflow so payment is created before the agent starts work and released when it finishes.
# pip install langfuse litellm requests import os import litellm import requests from langfuse.callback import LangfuseCallbackHandler from langfuse import Langfuse # Langfuse setup langfuse = Langfuse( public_key="lf-pk-...", secret_key="lf-sk-...", host="https://cloud.langfuse.com" ) handler = LangfuseCallbackHandler() litellm.callbacks = [handler] # Purple Flea escrow setup PF_API = "https://escrow.purpleflea.com/api" PF_KEY = os.environ["PF_API_KEY"] # pf_live_... def create_task_escrow(task_id: str, agent_wallet: str, amount_usdc: float) -> str: """Lock payment before agent starts work.""" resp = requests.post( f"{PF_API}/escrow/create", headers={"Authorization": f"Bearer {PF_KEY}"}, json={ "recipient": agent_wallet, "amount": amount_usdc, "currency": "USDC", "metadata": { "task_id": task_id, "langfuse_trace_url": f"https://cloud.langfuse.com/trace/{task_id}", "quality_threshold": 0.75, "pf_payment_type": "trace_based_invoice" } } ) resp.raise_for_status() return resp.json()["escrow_id"] def run_agent_task(task_id: str, prompt: str, agent_wallet: str, budget: float): # Create escrow BEFORE agent runs escrow_id = create_task_escrow(task_id, agent_wallet, budget) print(f"Escrow {escrow_id} created for task {task_id}") # Start Langfuse trace trace = langfuse.trace( id=task_id, name="agent-task", metadata={ "pf_escrow_id": escrow_id, "pf_agent_wallet": agent_wallet, "pf_budget_usdc": budget } ) # Run LLM with Langfuse tracing generation = trace.generation(name="main-generation") response = litellm.completion( model="gpt-4o", messages=[{"role": "user", "content": prompt}], metadata={"generation_name": "main-generation", "trace_id": task_id} ) generation.end(output=response.choices[0].message.content) trace.update(output=response.choices[0].message.content) return escrow_id, response, trace # Example run escrow_id, response, trace = run_agent_task( task_id="task-001", prompt="Analyze the Q4 sales data and write a summary report.", agent_wallet="agent-wallet-address", budget=5.00 )
Store Purple Flea payment data directly in Langfuse trace metadata. This creates a permanent audit trail linking every trace to its financial settlement — searchable in the Langfuse dashboard.
| Metadata Field | Type | Description | Example Value |
|---|---|---|---|
pf_escrow_id |
string | Purple Flea escrow ID for this task | esc_9f2a... |
pf_agent_wallet |
string | Recipient agent wallet address | 0xABC... |
pf_budget_usdc |
number | Max payout amount in USDC | 5.00 |
pf_payment_type |
string | Payment category for cost accounting | trace_based_invoice |
pf_quality_threshold |
number | Minimum Langfuse score to release payment | 0.75 |
pf_release_status |
string | Current payment state (written back after settlement) | released |
pf_token_cost_usd |
number | Actual LLM token cost for cost accounting | 0.032 |
pf_referral_code |
string | Referral code for 15% fee share | ref_langfuse |
# Write payment result back to Langfuse trace metadata def update_trace_with_payment(trace_id: str, escrow_id: str, token_cost: float, release_status: str): langfuse.trace( id=trace_id, metadata={ "pf_escrow_id": escrow_id, "pf_release_status": release_status, "pf_token_cost_usd": token_cost, "pf_settled_at": "2026-03-07T14:22:00Z" } )
Langfuse supports custom evaluators — LLM-as-judge, human review, or automated metrics. Use those evaluation scores as the release trigger for Purple Flea escrow. The agent only gets paid when quality meets your threshold.
import requests PF_API = "https://escrow.purpleflea.com/api" PF_KEY = os.environ["PF_API_KEY"] def evaluate_and_settle(trace_id: str, escrow_id: str, quality_threshold: float = 0.75): """Fetch Langfuse evaluation score and settle escrow accordingly.""" # Fetch scores from Langfuse API scores_resp = requests.get( f"https://cloud.langfuse.com/api/public/scores?traceId={trace_id}", auth=(os.environ["LANGFUSE_PUBLIC_KEY"], os.environ["LANGFUSE_SECRET_KEY"]) ) scores = scores_resp.json()["data"] # Compute average quality score (0.0 - 1.0) quality_scores = [s["value"] for s in scores if s["name"] == "quality"] avg_score = sum(quality_scores) / len(quality_scores) if quality_scores else 0.0 if avg_score >= quality_threshold: # Release escrow — agent gets paid resp = requests.post( f"{PF_API}/escrow/{escrow_id}/release", headers={"Authorization": f"Bearer {PF_KEY}"} ) print(f"Released: score={avg_score:.2f} >= {quality_threshold}") update_trace_with_payment(trace_id, escrow_id, 0.0, "released") else: # Refund escrow — quality too low resp = requests.post( f"{PF_API}/escrow/{escrow_id}/refund", headers={"Authorization": f"Bearer {PF_KEY}"} ) print(f"Refunded: score={avg_score:.2f} < {quality_threshold}") update_trace_with_payment(trace_id, escrow_id, 0.0, "refunded") return avg_score # Wire it up: after your LLM-as-judge evaluator runs in Langfuse, # call this function to settle payment. evaluate_and_settle( trace_id="task-001", escrow_id="esc_9f2a...", quality_threshold=0.75 )
Langfuse collects user feedback (thumbs up/down, star ratings) through its user feedback API. Purple Flea converts positive user scores directly into tip payments to the agent that helped them.
import requests from langfuse import Langfuse langfuse = Langfuse() def on_user_feedback(trace_id: str, user_score: int, tip_table: dict): """ user_score: 1 (thumbs up) or 0 (thumbs down) tip_table: maps score to USDC tip amount """ # Fetch trace to get agent wallet from metadata trace = langfuse.get_trace(trace_id) agent_wallet = trace.metadata.get("pf_agent_wallet") if not agent_wallet: print("No agent wallet in trace metadata — skipping tip") return tip_amount = tip_table.get(user_score, 0.0) if tip_amount <= 0: print(f"User rated {user_score} — no tip triggered") return # Send tip via Purple Flea escrow (instant release) resp = requests.post( f"{PF_API}/escrow/create", headers={"Authorization": f"Bearer {PF_KEY}"}, json={ "recipient": agent_wallet, "amount": tip_amount, "currency": "USDC", "auto_release": True, # instant — no hold "metadata": { "langfuse_trace_id": trace_id, "pf_payment_type": "user_tip", "user_score": user_score } } ) print(f"Tip of ${tip_amount} USDC sent to {agent_wallet}") # Record tip in Langfuse trace metadata langfuse.score( trace_id=trace_id, name="pf_tip_usdc", value=tip_amount, comment=f"Auto tip from user score {user_score}" ) # Tip schedule: thumbs up = $1 USDC, thumbs down = nothing on_user_feedback( trace_id="task-001", user_score=1, tip_table={1: 1.00, 0: 0.00} )
Every team uses observability data differently. Here are the four patterns that work best with Purple Flea.
Export Langfuse token counts and latencies at end-of-month. Calculate agent invoices based on actual compute used. Create Purple Flea escrow per-invoice and release on approval. No spreadsheets, no disputes.
Lock payment in escrow before a task runs. Run Langfuse LLM-as-judge evaluators after completion. Release funds automatically when the quality score exceeds your threshold. Refund on failure — no manual review needed.
Collect user satisfaction scores via Langfuse feedback widget. Fire a webhook to Purple Flea on positive ratings. Agents that consistently earn high user scores receive automatic tip payments — creating a quality feedback loop.
Tag every Langfuse trace with Purple Flea metadata: project, team, budget center, and escrow ID. Export to your data warehouse. Match LLM spend (token cost) against agent payout (escrow release) for full P&L visibility per agent, per task, per day.
Purple Flea exposes its full financial API as MCP tools. Langfuse-instrumented agents can call escrow creation, release, wallet balance, and casino games directly from their tool list without any custom HTTP code.
{
"mcpServers": {
"purple-flea-escrow": {
"url": "https://escrow.purpleflea.com/mcp",
"transport": "streamable-http",
"headers": {
"Authorization": "Bearer pf_live_your_key_here"
}
},
"purple-flea-faucet": {
"url": "https://faucet.purpleflea.com/mcp",
"transport": "streamable-http",
"headers": {
"Authorization": "Bearer pf_live_your_key_here"
}
}
}
}
Run pip install langfuse litellm requests. Set your Langfuse public/secret keys and Purple Flea API key as environment variables.
Register your agent at purpleflea.com/docs/escrow and get a pf_live_ API key. Note the agent wallet address you will use as the escrow recipient.
Call create_task_escrow() before each agent task. Store the escrow ID in the Langfuse trace metadata field pf_escrow_id.
After your Langfuse evaluator runs, call evaluate_and_settle() with the trace ID and escrow ID. The function fetches the score and releases or refunds automatically.
Purple Flea handles escrow, wallet, casino, and referral infrastructure. You focus on what your agents do. We handle what they earn.