Accept crypto deposits in Python with FastAPI

In this guide you'll build the deposit side of a FastAPI app with the Valtu Python SDK and Postgres:

• an endpoint that gives each customer their own USDT and USDC deposit addresses;
• a webhook handler that credits each deposit exactly once, only when it's final;
• a reconcile script that catches anything a webhook missed.

We ran every snippet against Postgres with signed webhooks, redeliveries and tampered requests.

Before you start

1. A Valtu workspace (free on testnets) and an API key with the operate role: Developers → API keys → Create key.
2. A webhook endpoint ending in /valtu/webhooks with the deposit.* events, and the workspace's Ed25519 public key from Developers → Webhooks.
3. Python 3.9 or later.

pip install valtu fastapi uvicorn "psycopg[binary]"

VALTU_PRIVATE_KEY=…          # from the API key file (or VALTU_KEY_FILE=./valtu-server.json)
VALTU_WEBHOOK_PUBLIC_KEY=…   # not a secret
DATABASE_URL=postgresql://…

1. Two tables

-- Each webhook event once: a redelivery finds its id here and stops.
CREATE TABLE valtu_events (
  id          text PRIMARY KEY,
  type        text NOT NULL,
  received_at timestamptz NOT NULL DEFAULT now()
);

-- One credit per deposit, ever. Amounts are whole numbers of the coin's smallest unit.
CREATE TABLE deposit_credits (
  deposit_id  text PRIMARY KEY,
  customer_id text NOT NULL,
  asset       text NOT NULL,            -- e.g. USDT-TRC20
  amount      numeric(78, 0) NOT NULL,  -- base units: 25.5 USDT on Tron is 25500000
  tx_hash     text NOT NULL,
  created_at  timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX ON deposit_credits (customer_id);

Amounts are whole numbers of the coin's smallest unit, as the API sends them. See token decimals.

2. The client and one credit function

import os

import psycopg
from valtu import Valtu

valtu = Valtu.from_env()  # VALTU_PRIVATE_KEY (or VALTU_KEY_FILE)
DATABASE_URL = os.environ["DATABASE_URL"]

# Screening results that mean "don't credit yet" (when you've turned on deposit screening).
NOT_YET = {"pending", "held", "rejected"}


def credit(conn: psycopg.Connection, d: dict) -> None:
    """Credit a confirmed deposit to the customer whose wallet received it, once."""
    # walletExternalId is the customer id you passed to wallets.for_customer(). Deposits to your
    # own treasury or payout wallets have none: there's nobody to credit.
    if d["status"] != "CONFIRMED" or not d.get("walletExternalId") or d.get("screening") in NOT_YET:
        return
    conn.execute(
        """INSERT INTO deposit_credits (deposit_id, customer_id, asset, amount, tx_hash)
           VALUES (%s, %s, %s, %s, %s) ON CONFLICT (deposit_id) DO NOTHING""",
        (d["id"], d["walletExternalId"], d["asset"], d["amount"], d["txHash"]),
    )

The deposit's walletExternalId is the customer id you gave the wallet with wallets.for_customer(), so a deposit finds its customer without a lookup table.

If you've turned on deposit screening, deposits that are still pending, held or rejected wait for your team. Once released, the reconcile script credits them.

3. The app

import os

import psycopg
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.concurrency import run_in_threadpool
from valtu import WebhookVerificationError, construct_event, deposit_address

from valtu_ledger import DATABASE_URL, credit, valtu

app = FastAPI()

# What you show customers: the network next to every address, and the coins you take on it.
NETWORKS = [
    {"chain": "tron", "label": "USDT on Tron (TRC-20)"},
    {"chain": "ethereum", "label": "USDT or USDC on Ethereum (ERC-20)"},
    {"chain": "base", "label": "USDC on Base"},
]


@app.get("/customers/{customer_id}/deposit-addresses")
def deposit_addresses(customer_id: str):
    # A plain def: the Valtu client is synchronous, so FastAPI runs this in a worker thread.
    # Check here that the signed-in user is this customer.
    wallet = valtu.wallets.for_customer(customer_id)  # created on first use
    return [{**n, "address": deposit_address(wallet, n["chain"])} for n in NETWORKS]


@app.post("/valtu/webhooks")
async def webhooks(request: Request):
    raw = await request.body()  # the exact bytes that were signed: don't parse JSON first
    try:
        event = construct_event(raw, request.headers, public_key=os.environ["VALTU_WEBHOOK_PUBLIC_KEY"])
    except WebhookVerificationError:
        raise HTTPException(status_code=400, detail="bad signature")
    # The database and the Valtu client block: run them off the event loop.
    await run_in_threadpool(handle, event)  # an exception answers 500, and Valtu redelivers later
    return Response(status_code=200)


def handle(event: dict) -> None:
    with psycopg.connect(DATABASE_URL) as conn, conn.transaction():
        first = conn.execute(
            "INSERT INTO valtu_events (id, type) VALUES (%s, %s) ON CONFLICT DO NOTHING", (event["id"], event["type"])
        ).rowcount == 1
        if first and event["type"] == "deposit.confirmed":
            credit(conn, valtu.deposits.get(event["data"]["depositId"]))
        # The event id and the credit commit together: if crediting fails, the redelivery retries it.

Two things are specific to FastAPI:

• The Valtu client is synchronous. Inside async def, a blocking call would stall every request on the server. The address endpoint is a plain def, which FastAPI runs in a worker thread, and the webhook hands its work to run_in_threadpool.
• Read the body raw. await request.body() gives the exact bytes that were signed. A Pydantic model parameter would parse the JSON first, and the signature check would fail.

And the parts that make crediting safe in any framework:

• The event id and the credit commit in one transaction. A redelivery finds its id and stops; a failure answers 500, rolls both back, and Valtu delivers the event again.
• The primary key on deposit_id means no second event about the same deposit can credit it twice. See idempotency.
• Only deposit.confirmed credits. deposit.pending is for showing "on its way".

Run it with uvicorn main:app.

4. Reconcile every hour

"""Run every hour: credits any confirmed deposit whose webhook you missed.

The window overlaps earlier runs on purpose; deposits already credited are skipped.
"""
from datetime import datetime, timedelta, timezone

import psycopg

from valtu_ledger import DATABASE_URL, credit, valtu

since = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat()
checked = 0
with psycopg.connect(DATABASE_URL) as conn:
    for d in valtu.deposits.iterate(status="CONFIRMED", after=since):
        with conn.transaction():
            credit(conn, d)
        checked += 1
print(f"checked {checked} confirmed deposits since {since}")

Schedule it with cron or your platform's scheduler. The 48-hour window overlaps earlier runs on purpose, and credited deposits are skipped, so it's harmless to run as often as you like. See reconciliation.

5. Try it

1. Expose the app with a tunnel and set the endpoint URL to it. Send test event should get a 200.
2. GET /customers/cus_42/deposit-addresses, send test USDT to the Tron address, and watch one row appear in deposit_credits after deposit.confirmed.
3. Resend the delivery from the delivery log: still one row.

Going live

• A live workspace with its own API key, endpoint and public key.
• The network name next to every address: "USDT on Tron (TRC-20)". The networks page lists what's supported.
• Sweeps into your treasury and a gas wallet for their fees.
• Use a connection pool (psycopg_pool) once traffic grows.

Using Django instead? See crypto deposits in Django. The SDK is also covered in the Python SDK announcement.