| """ |
| DPX Protocol — Hugging Face Space |
| Live demo of the DPX settlement infrastructure via MCP-callable endpoints. |
| github.com/untitledfinancial/dpx-mcp · docs.untitledfinancial.com |
| """ |
|
|
| import gradio as gr |
| import httpx |
| import json |
| from datetime import datetime |
|
|
| STABILITY_URL = "https://stability.untitledfinancial.com" |
| ESG_URL = "https://esg.untitledfinancial.com" |
| AGENT_URL = "https://agent.untitledfinancial.com" |
| TIMEOUT = 12.0 |
|
|
|
|
| def _get(url: str) -> dict: |
| with httpx.Client(timeout=TIMEOUT) as client: |
| r = client.get(url, headers={"User-Agent": "DPX-HF-Space/1.0"}) |
| r.raise_for_status() |
| return r.json() |
|
|
|
|
| |
|
|
| def get_quote(amount_usd: float, has_fx: bool, esg_score: float) -> str: |
| try: |
| params = f"amountUsd={amount_usd}&hasFx={'true' if has_fx else 'false'}&esgScore={int(esg_score)}" |
| data = _get(f"{AGENT_URL}/quote?{params}") |
| lines = [ |
| f"**Quote ID:** `{data.get('quoteId', 'N/A')}`", |
| f"**Amount:** ${amount_usd:,.2f} USD", |
| f"**Net amount (after fees):** ${data.get('netAmountUsd', 0):,.2f} USD", |
| "", |
| "**Fee breakdown:**", |
| ] |
| fees = data.get("fees", {}) |
| for key in ["core", "fx", "esg", "license", "total"]: |
| f = fees.get(key, {}) |
| if f: |
| lines.append(f"- {key.capitalize()}: {f.get('pct', 0):.3f}% (${f.get('usd', 0):,.2f})") |
| lines += [ |
| "", |
| f"**Oracle status:** {data.get('oracleStatus', 'N/A')} (score: {data.get('oracleScore', 'N/A')})", |
| f"**Volume tier:** {data.get('tier', 'N/A')}", |
| f"**Quote expires:** {data.get('expiresAt', 'N/A')}", |
| "", |
| f"**AI reasoning:** {data.get('reasoning', 'N/A')}", |
| ] |
| return "\n".join(lines) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
|
|
| |
|
|
| def get_oracle_status() -> str: |
| try: |
| data = _get(f"{STABILITY_URL}/reliability") |
| lines = [ |
| f"**Status:** {data.get('status', 'N/A')}", |
| f"**Stability score:** {data.get('stabilityScore', data.get('score', 'N/A'))} / 100", |
| f"**Recommendation:** {data.get('recommendation', 'N/A')}", |
| f"**Peg deviation:** {data.get('pegDeviation', 'N/A')} bps", |
| f"**Outlook:** {data.get('outlook', 'N/A')}", |
| "", |
| f"**AI reasoning:** {data.get('reasoning', data.get('briefing', 'N/A'))}", |
| "", |
| f"*Last updated: {data.get('timestamp', datetime.utcnow().isoformat())}*", |
| ] |
| return "\n".join(lines) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
|
|
| def get_oracle_signals() -> str: |
| try: |
| data = _get(f"{STABILITY_URL}/oracle-status") |
| signals = data.get("signals", {}) |
| if not signals: |
| return json.dumps(data, indent=2) |
| lines = [ |
| f"**Composite score:** {data.get('score', 'N/A')} · **Status:** {data.get('status', 'N/A')}", |
| f"**Chaos regime:** {'⚠️ YES' if data.get('chaosRegime') else 'No'}", |
| "", |
| "**Signal layers (0–100, higher = more stable):**", |
| ] |
| for sig, val in signals.items(): |
| bar = "█" * int(val / 10) + "░" * (10 - int(val / 10)) |
| lines.append(f"- {sig:<16} {bar} {val:.0f}") |
| alerts = data.get("alerts", []) |
| if alerts: |
| lines += ["", "**Active alerts:**"] + [f"- {a}" for a in alerts] |
| briefing = data.get("briefing", "") |
| if briefing: |
| lines += ["", f"**Intelligence briefing:** {briefing}"] |
| return "\n".join(lines) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
|
|
| |
|
|
| def get_esg_score(address: str) -> str: |
| try: |
| addr = address.strip() or "default" |
| path = f"/esg-score/{addr}" if addr != "default" else "/esg-score" |
| data = _get(f"{ESG_URL}{path}") |
| lines = [ |
| f"**Address:** `{data.get('address', addr)}`", |
| f"**Composite ESG score:** {data.get('esgScore', data.get('score', 'N/A'))} / 100", |
| "", |
| "**Sub-scores:**", |
| f"- Environmental: {data.get('environmental', 'N/A')}", |
| f"- Social: {data.get('social', 'N/A')}", |
| f"- Governance: {data.get('governance', 'N/A')}", |
| "", |
| f"**ESG fee at settlement:** {data.get('feePct', 'N/A')}%", |
| f"**Tier:** {data.get('tier', 'N/A')}", |
| f"**Last updated:** {data.get('updatedAt', 'N/A')}", |
| ] |
| sources = data.get("sources", []) |
| if sources: |
| lines += ["", f"**Data sources:** {', '.join(sources)}"] |
| return "\n".join(lines) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
|
|
| |
|
|
| def get_fx_corridors() -> str: |
| try: |
| data = _get(f"{STABILITY_URL}/fx-settlement") |
| lines = [ |
| f"**Overall risk:** {data.get('overallRisk', 'N/A')}", |
| ] |
| dxy = data.get("dxy", {}) |
| if dxy: |
| lines += [ |
| f"**DXY:** {dxy.get('value', 'N/A')} · Regime: {dxy.get('regime', 'N/A')}", |
| "", |
| ] |
| corridors = data.get("corridors", []) |
| if corridors: |
| lines.append("**Per-corridor risk:**") |
| for c in corridors: |
| pair = c.get("pair", "?") |
| advice = c.get("advice", "?") |
| risk = c.get("risk", "?") |
| spot = c.get("spotRate", "") |
| spot_s = f" ({spot})" if spot else "" |
| lines.append(f"- **{pair}**{spot_s}: {risk} → {advice}") |
| summary = data.get("executiveSummary", "") |
| if summary: |
| lines += ["", f"**Summary:** {summary}"] |
| actions = data.get("recommendedActions", []) |
| if actions: |
| lines += ["", "**Recommended actions:**"] + [f"- {a}" for a in actions] |
| return "\n".join(lines) |
| except Exception as e: |
| return f"Error: {e}" |
|
|
|
|
| |
|
|
| with gr.Blocks( |
| title="DPX Protocol — Live Demo", |
| theme=gr.themes.Base(primary_hue="slate"), |
| css=".gr-button-primary { background: #18181b !important; }", |
| ) as demo: |
| gr.Markdown( |
| """ |
| # DPX Protocol — Live Settlement Infrastructure |
| |
| Compliance-grade stablecoin settlement rail. Any agent can discover, price, and execute cross-border payments autonomously — no onboarding, no API key, no human step. |
| |
| **Contracts live on Base mainnet · [docs.untitledfinancial.com](https://docs.untitledfinancial.com) · [MCP server](https://smithery.ai/server/@untitledfinancial/dpx-mcp)** |
| """ |
| ) |
|
|
| with gr.Tabs(): |
|
|
| with gr.TabItem("Settlement Quote"): |
| gr.Markdown("Get a binding fee quote. All calculations run against the live Stability Oracle.") |
| with gr.Row(): |
| amount = gr.Number(label="Amount (USD)", value=500000, minimum=1) |
| has_fx = gr.Checkbox(label="Cross-currency (FX)", value=True) |
| esg_sl = gr.Slider(label="ESG score (0–100)", minimum=0, maximum=100, value=75, step=1) |
| quote_btn = gr.Button("Get quote", variant="primary") |
| quote_out = gr.Markdown() |
| quote_btn.click(get_quote, inputs=[amount, has_fx, esg_sl], outputs=quote_out) |
|
|
| with gr.TabItem("Oracle Status"): |
| gr.Markdown("Live macro stability assessment. Gates every settlement — UNSTABLE halts execution automatically.") |
| with gr.Row(): |
| oracle_btn = gr.Button("Check stability", variant="primary") |
| oracle_sig_btn = gr.Button("Full signal layers") |
| oracle_out = gr.Markdown() |
| oracle_btn.click(get_oracle_status, outputs=oracle_out) |
| oracle_sig_btn.click(get_oracle_signals, outputs=oracle_out) |
|
|
| with gr.TabItem("ESG Score"): |
| gr.Markdown("Score a counterparty wallet for ESG risk. Required for EU SFDR/CSRD compliance. Updated hourly from 6 institutional sources.") |
| addr_in = gr.Textbox(label="Wallet address (0x...) — leave blank for protocol default", placeholder="0x...") |
| esg_btn = gr.Button("Get ESG score", variant="primary") |
| esg_out = gr.Markdown() |
| esg_btn.click(get_esg_score, inputs=addr_in, outputs=esg_out) |
|
|
| with gr.TabItem("FX Corridors"): |
| gr.Markdown("Per-pair execution risk for 10 major currency corridors. SETTLE_NOW / DELAY / AVOID recommendations updated hourly.") |
| fx_btn = gr.Button("Get FX corridor status", variant="primary") |
| fx_out = gr.Markdown() |
| fx_btn.click(get_fx_corridors, outputs=fx_out) |
|
|
| gr.Markdown( |
| """ |
| --- |
| **Integration:** `npx @untitledfinancial/dpx-mcp` · Listed on Anthropic MCP registry, Smithery, mcp.so, npm |
| **Agentic payments:** x402 micropayment standard on Base mainnet — agents pay per call in USDC, no API key required |
| **Compliance:** FATF R16 · MiCA · GENIUS Act · SFDR/CSRD · Basel III Group 1b |
| """ |
| ) |
|
|
| if __name__ == "__main__": |
| demo.launch() |
|
|
