hjess/tink-demo
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Wiki Activity

feat: Tink open banking demo — 6-step API walkthrough

Browse Source 
Demonstrates the full Tink integration flow for open banking:
  Step 1 — Client credentials auth (app token)
  Step 2 — Create Tink user with external_user_id
  Step 3 — Connect bank via Tink Link OAuth redirect
  Step 4 — List accounts (v2 endpoint)
  Step 5 — List transactions (v2 endpoint, cursor pagination)
  Step 6 — Webhooks (register endpoint, receive events)

Built with Python / FastAPI + Jinja2 templates.
Each step shows live JSON responses, cURL examples and API version badges.
Includes server-side token store (prevents session cookie overflow),
asyncio lock on OAuth callback, and demo mode with realistic mock data.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Henrik Jess Nielsen
2026-05-23 02:08:27 +02:00
commit ab591be464
23 changed files with 2477 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

644
src/routes/demo.py Normal file
View File

@@ -0,0 +1,644 @@
"""
Demo routes — each /demo/step/N page shows one Tink API call
with the live JSON response and a curl example.
"""
import json
import uuid
import secrets
from typing import Optional
from fastapi import APIRouter, Request, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from src.tink.client import TinkClient
from src.config import get_settings
from src import demo_data
router = APIRouter()
templates = Jinja2Templates(directory="src/templates")
def _client(log_cb=None) -> TinkClient:
s = get_settings()
return TinkClient(
client_id=s.tink_client_id,
client_secret=s.tink_client_secret,
redirect_uri=s.tink_redirect_uri,
api_base=s.tink_api_base,
link_base=s.tink_link_base,
on_request=log_cb,
)
def _session(request: Request) -> dict:
return request.session.setdefault("demo", {})
# ---------------------------------------------------------------------------
# Server-side token store — keeps JWTs OUT of the session cookie
# (cookie limit is 4KB; two JWTs alone are ~1.3KB before base64 overhead)
# ---------------------------------------------------------------------------
_token_store: dict[str, dict] = {} # sid → {"app_token": str, "user_token": str}
def _get_sid(sess: dict) -> str:
"""Return (creating if needed) a stable session ID stored in the cookie."""
if "sid" not in sess:
sess["sid"] = str(uuid.uuid4())
return sess["sid"]
def _store_token(sess: dict, key: str, value: str) -> None:
"""Save a JWT in the server-side store instead of the cookie."""
sid = _get_sid(sess)
_token_store.setdefault(sid, {})[key] = value
def _load_token(sess: dict, key: str, default: str = "") -> str:
"""Read a JWT from the server-side store."""
sid = sess.get("sid", "")
return _token_store.get(sid, {}).get(key, default)
def _ctx(request: Request, extra: dict) -> dict:
"""Base template context — always includes session_customer."""
sess = _session(request)
return {"request": request, "session_customer": sess.get("external_user_id", ""), **extra}
def _logger(sess: dict):
"""Returns a callback that appends log entries to sess['api_log']."""
def cb(entry: dict):
log = sess.setdefault("api_log", [])
log.append(entry)
# keep last 50 entries
if len(log) > 50:
sess["api_log"] = log[-50:]
return cb
# ---------------------------------------------------------------------------
# Landing
# ---------------------------------------------------------------------------
@router.get("/", response_class=HTMLResponse)
async def index(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@router.get("/demo/reset")
async def reset_demo(request: Request):
"""Clear all demo session state and restart from Step 1."""
sess = _session(request)
sid = sess.get("sid", "")
if sid:
_token_store.pop(sid, None)
request.session.pop("demo", None)
return RedirectResponse("/demo/step/1")
@router.get("/demo/debug-session")
async def debug_session(request: Request):
"""Show current session keys (debug only)."""
sess = _session(request)
safe = {
k: (v[:20] + "…" if isinstance(v, str) and len(v) > 20 else v)
for k, v in sess.items()
if k != "api_log"
}
safe["api_log_count"] = len(sess.get("api_log", []))
safe["cookie_size_bytes"] = len(str(request.session))
return safe
# ---------------------------------------------------------------------------
# Step 1 — Authenticate (client credentials)
# ---------------------------------------------------------------------------
@router.get("/demo/step/1", response_class=HTMLResponse)
async def step1(request: Request):
sess = _session(request)
client = _client(log_cb=_logger(sess))
s = get_settings()
error = None
result = None
curl_example = (
f"curl -X POST https://api.tink.com/api/v1/oauth/token \\\n"
f" -d 'client_id={s.tink_client_id}' \\\n"
f" -d 'client_secret=***' \\\n"
f" -d 'grant_type=client_credentials' \\\n"
f" -d 'scope=user:create,authorization:grant'"
)
try:
result = await client.get_app_token(scope="user:create,authorization:grant")
_store_token(sess, "app_token", result["access_token"])
except Exception as e:
error = str(e)
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 1,
"title": "Authenticate",
"subtitle": "Client Credentials Flow",
"endpoint": "POST /api/v1/oauth/token",
"api_version": "v1",
"description": "Vi starter med at hente et app-level access token via Client Credentials flow. Dette token bruges til at oprette brugere.",
"curl_example": curl_example,
"result": result,
"error": error,
"next_step": 2,
"prev_step": None,
}))
# ---------------------------------------------------------------------------
# Step 2 — Create User
# ---------------------------------------------------------------------------
@router.get("/demo/step/2", response_class=HTMLResponse)
async def step2_get(request: Request):
sess = _session(request)
s = get_settings()
app_token = _load_token(sess, "app_token")
error = None
if not app_token:
error = "Mangler app token — gå tilbage til Step 1 og kør authentication først."
# If user already created this session, skip the form
existing_user_id = sess.get("user_id", "")
existing_external_id = sess.get("external_user_id", "")
return templates.TemplateResponse("step2.html", _ctx(request, {
"step": 2,
"error": error,
"existing_user_id": existing_user_id,
"existing_external_id": existing_external_id,
"app_token_ok": bool(app_token),
"next_step": 3,
"prev_step": 1,
}))
@router.post("/demo/step/2", response_class=HTMLResponse)
async def step2_post(request: Request,
customer_name: str = Form(default=""),
market: str = Form(default="DK")):
sess = _session(request)
s = get_settings()
app_token = _load_token(sess, "app_token")
error = None
result = None
# Build external_user_id — always unique per run (simulates real customer UUID)
short_id = secrets.token_hex(3) # 6-char hex, e.g. "a3f9c1"
if customer_name.strip():
import re
slug = customer_name.strip().lower().replace(" ", "-")
slug = re.sub(r"[^a-z0-9\-]", "", slug)
external_user_id = f"moneycapp-{slug}-{short_id}"
else:
external_user_id = f"moneycapp-{short_id}"
body = json.dumps({"external_user_id": external_user_id, "market": market, "locale": "da_DK"}, indent=2)
curl_example = (
f"curl -X POST https://api.tink.com/api/v1/user/create \\\n"
f" -H 'Authorization: Bearer $APP_TOKEN' \\\n"
f" -H 'Content-Type: application/json' \\\n"
f" -d '{body}'"
)
if not app_token:
error = "Mangler app token — gå tilbage til Step 1."
else:
try:
client = _client(log_cb=_logger(sess))
result = await client.create_user(app_token, external_user_id, market=market)
sess["user_id"] = result.get("user_id", "")
sess["external_user_id"] = external_user_id
sess["user_market"] = market
# New user — clear any stale tokens from a previous user
sess.pop("user_token", None)
sess.pop("demo_mode", None)
except Exception as e:
error = str(e)
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 2,
"title": "Opret Bruger",
"subtitle": f"Kunde: {external_user_id}",
"endpoint": "POST /api/v1/user/create",
"api_version": "v1",
"description": f"Oprettet Tink-bruger med <code class='text-violet-300 font-mono'>tink_external_ref</code> = <code class='text-violet-300 font-mono'>{external_user_id}</code> — MoneyCapp's reference til kunden i Tink. Tink returnerer et <code class='text-violet-300 font-mono'>user_id</code> som bruges i næste kald.",
"curl_example": curl_example,
"result": result,
"error": error,
"next_step": 3,
"prev_step": 1,
}))
# ---------------------------------------------------------------------------
# Step 3 — Connect Bank (Tink Link redirect)
# ---------------------------------------------------------------------------
CONSOLE_CALLBACK = "https://console.tink.com/callback"
@router.get("/demo/step/3", response_class=HTMLResponse)
async def step3_get(request: Request):
sess = _session(request)
s = get_settings()
client = _client(log_cb=_logger(sess))
error = None
credentials = None
cb_success = request.query_params.get("cb_success")
cb_error = request.query_params.get("cb_error")
user_id = sess.get("user_id", "")
app_token = _load_token(sess, "app_token")
grant_result = None
# Call authorization-grant/delegate to demonstrate the API and get a code.
# We show this in the response viewer so the audience sees it works.
# The actual Tink Link flow uses anonymous mode (sandbox limitation),
# but in production you'd pass authorization_code in the URL.
if user_id and app_token and not _load_token(sess, "user_token"):
try:
grant_result = await client.get_authorization_grant_token(
app_token=app_token,
user_id=user_id,
scope="accounts:read,transactions:read,credentials:read",
)
except Exception as e:
error = str(e)
# Always use anonymous flow for the actual Tink Link URL — works in sandbox.
# Production apps would add authorization_code=<code> instead of scope.
tink_link_url = client.get_tink_link_url(market=sess.get("user_market", "DK"))
dev_tink_link_url = client.get_tink_link_url(
market=sess.get("user_market", "DK"),
redirect_uri_override=CONSOLE_CALLBACK,
)
# Demo mode: auto-mark as connected with mock data
if s.demo_mode and not _load_token(sess, "user_token"):
_store_token(sess, "user_token", "demo-mode-token")
sess["demo_mode"] = True
# Check if already connected (returning from callback)
user_token = _load_token(sess, "user_token")
if user_token:
if sess.get("demo_mode"):
credentials = demo_data.MOCK_CREDENTIALS
cb_success = cb_success or "demo"
else:
try:
credentials = await client.list_credentials(user_token)
except Exception as e:
if "403" in str(e):
credentials = {"note": "credentials:read scope ikke tildelt — brug accounts:read i stedet"}
else:
error = str(e)
external_user_id = sess.get("external_user_id", "")
uid_display = user_id or "$USER_ID"
grant_code = (grant_result or {}).get("code", "$AUTHORIZATION_CODE")
curl_example = (
"# Step 1: Generer authorization_code for din specifikke bruger\n"
"curl -X POST https://api.tink.com/api/v1/oauth/authorization-grant/delegate \\\n"
" -H 'Authorization: Bearer $APP_TOKEN' \\\n"
" -d 'actor_client_id=$CLIENT_ID' \\\n"
f" -d 'user_id={uid_display}' \\\n"
" -d 'scope=accounts:read,transactions:read'\n"
"# → { \"code\": \"AUTHORIZATION_CODE\" }\n\n"
"# Step 2: Byg Tink Link URL med authorization_code (binder bank til din bruger)\n"
"https://link.tink.com/1.0/transactions/connect-accounts\n"
" ?client_id=$CLIENT_ID\n"
" &redirect_uri=$REDIRECT_URI\n"
f" &authorization_code={grant_code}\n\n"
"# Step 3: Callback → exchange code for user token\n"
"curl -X POST https://api.tink.com/api/v1/oauth/token \\\n"
" -d 'grant_type=authorization_code' \\\n"
" -d 'client_id=$CLIENT_ID' \\\n"
" -d 'client_secret=$CLIENT_SECRET' \\\n"
" -d 'code=$CALLBACK_CODE'"
)
if grant_result:
description = (
f"authorization-grant/delegate kaldt for bruger <code class='text-violet-300 font-mono'>{external_user_id}</code> — "
f"returnerede <code class='text-violet-300 font-mono'>code</code> som vist nedenfor. "
f"I produktion sendes denne code med i Tink Link URL som <code class='text-violet-300 font-mono'>authorization_code</code>."
)
elif user_id:
description = (
"Åbn Tink Link → vælg <b>Tink Demo Bank</b> → log ind med testbruger. "
"Klik <b>Vis testbrugere</b> for at se login til DK, SE, NO, FI, DE m.fl."
)
else:
description = "Gå til Step 2 for at oprette en bruger, klik derefter her for at tilslutte banken."
# Tink Demo Bank test users per market
# Source: https://docs.tink.com/resources/tutorials/test-your-integration-with-demo-bank
demo_bank_users = [
{"market": "🇩🇰 DK", "username": "u04877810", "password": "vxw774", "otp": "1234", "scenario": "Succes"},
{"market": "🇩🇰 DK", "username": "u92721594", "password": "nbs589", "otp": "", "scenario": "Auth-fejl"},
{"market": "🇸🇪 SE", "username": "u59803783", "password": "hwj858", "otp": "1234", "scenario": "Succes"},
{"market": "🇸🇪 SE", "username": "u91817276", "password": "cft248", "otp": "", "scenario": "Auth-fejl"},
{"market": "🇳🇴 NO", "username": "u24765398", "password": "xjf459", "otp": "1234", "scenario": "Succes"},
{"market": "🇫🇮 FI", "username": "u19283746", "password": "zkm291", "otp": "1234", "scenario": "Succes"},
{"market": "🇩🇪 DE", "username": "u38471920", "password": "bvp103", "otp": "1234", "scenario": "Succes"},
{"market": "🇬🇧 GB", "username": "u72910483", "password": "qrt567", "otp": "1234", "scenario": "Succes"},
{"market": "🇳🇱 NL", "username": "u56473829", "password": "lmn482", "otp": "1234", "scenario": "Succes"},
{"market": "🇫🇷 FR", "username": "u84920173", "password": "pqs736", "otp": "1234", "scenario": "Succes"},
]
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 3,
"title": "Tilslut Bank",
"subtitle": "Tink Link — Bank Connection",
"endpoint": "POST /api/v1/oauth/authorization-grant/delegate",
"api_version": "v1",
"description": description,
"curl_example": curl_example,
"result": grant_result or credentials,
"tink_link_url": tink_link_url,
"dev_tink_link_url": dev_tink_link_url,
"demo_bank_users": demo_bank_users,
"error": error or (f"Callback fejl: {cb_error}" if cb_error else None),
"cb_success": cb_success,
"next_step": 4,
"prev_step": 2,
}))
@router.post("/demo/step/3", response_class=HTMLResponse)
async def step3_post(request: Request, code: str = Form(...)):
"""Manual code entry — exchange a code obtained via console.tink.com/callback."""
sess = _session(request)
try:
client = _client(log_cb=_logger(sess))
tokens = await client.exchange_code_for_token(
code=code.strip(),
redirect_uri=CONSOLE_CALLBACK,
)
_store_token(sess, "user_token", tokens.get("access_token", ""))
return RedirectResponse("/demo/step/3?cb_success=1", status_code=303)
except Exception as e:
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}", status_code=303)
@router.get("/callback", response_class=HTMLResponse)
async def tink_callback(request: Request, code: Optional[str] = None,
error: Optional[str] = None,
credentials_id: Optional[str] = None):
"""Tink Link OAuth callback — exchange code for user token."""
sess = _session(request)
print(f"[CALLBACK] code={code!r} error={error!r} session_keys={list(sess.keys())}")
if error:
print(f"[CALLBACK] Tink returned error: {error}")
return RedirectResponse(f"/demo/step/3?error={error}")
if code:
try:
s = get_settings()
print(f"[CALLBACK] Exchanging code, redirect_uri={s.tink_redirect_uri!r}")
client = _client(log_cb=_logger(sess))
tokens = await client.exchange_code_for_token(
code, redirect_uri=s.tink_redirect_uri
)
print(f"[CALLBACK] Token response keys: {list(tokens.keys())}")
user_token = tokens.get("access_token", "")
if not user_token:
print(f"[CALLBACK] ERROR: access_token missing, got: {tokens}")
return RedirectResponse(
f"/demo/step/3?cb_error=Token+exchange+ok+but+access_token+missing+in+response",
status_code=303,
)
_store_token(sess, "user_token", user_token)
print(f"[CALLBACK] SUCCESS — user_token saved ({len(user_token)} chars)")
return RedirectResponse("/demo/step/3?cb_success=1", status_code=303)
except Exception as e:
import traceback
print(f"[CALLBACK] EXCEPTION: {e}\n{traceback.format_exc()}")
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}", status_code=303)
print(f"[CALLBACK] No code — bare redirect to step 3")
return RedirectResponse("/demo/step/3", status_code=303)
# ---------------------------------------------------------------------------
# Step 4 — Accounts (v2)
# ---------------------------------------------------------------------------
@router.get("/demo/step/4", response_class=HTMLResponse)
async def step4(request: Request):
sess = _session(request)
s = get_settings()
user_token = _load_token(sess, "user_token")
error = None
result = None
is_demo = sess.get("demo_mode") or s.demo_mode
curl_example = (
"curl https://api.tink.com/data/v2/accounts \\\n"
" -H 'Authorization: Bearer $USER_TOKEN'"
)
if not user_token and not is_demo:
error = "Mangler user token — tilslut en bank i Step 3 først."
elif is_demo:
result = demo_data.MOCK_ACCOUNTS
else:
try:
client = _client(log_cb=_logger(sess))
result = await client.list_accounts(user_token)
# Fall back to demo data if no accounts connected yet
if not result.get("accounts"):
result = demo_data.MOCK_ACCOUNTS
sess["demo_mode"] = True
except Exception as e:
error = str(e)
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 4,
"title": "Konti",
"subtitle": "Account List with Balances",
"endpoint": "GET /data/v2/accounts",
"api_version": "v2 ✦",
"description": "Henter brugerens konti via det nye v2 data endpoint. Returnerer account type, balance, currency og IBAN.",
"curl_example": curl_example,
"result": result,
"error": error,
"is_demo": is_demo or (result == demo_data.MOCK_ACCOUNTS),
"next_step": 5,
"prev_step": 3,
}))
# ---------------------------------------------------------------------------
# Step 5 — Transactions (v2)
# ---------------------------------------------------------------------------
@router.get("/demo/step/5", response_class=HTMLResponse)
async def step5(request: Request, account_id: Optional[str] = None):
sess = _session(request)
s = get_settings()
user_token = _load_token(sess, "user_token")
is_demo = sess.get("demo_mode") or s.demo_mode
error = None
result = None
curl_example = (
"curl 'https://api.tink.com/data/v2/transactions?pageSize=25' \\\n"
" -H 'Authorization: Bearer $USER_TOKEN'"
)
if not user_token and not is_demo:
error = "Mangler user token — tilslut en bank i Step 3 først."
elif is_demo:
result = demo_data.MOCK_TRANSACTIONS
else:
try:
client = _client(log_cb=_logger(sess))
result = await client.list_transactions(user_token, account_id=account_id)
if not result.get("transactions"):
result = demo_data.MOCK_TRANSACTIONS
sess["demo_mode"] = True
except Exception as e:
error = str(e)
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 5,
"title": "Transaktioner",
"subtitle": "Transaction History",
"endpoint": "GET /data/v2/transactions",
"api_version": "v2 ✦",
"description": "Henter transaktioner via v2 endpoint med paginering. Returnerer amount, description, status (BOOKED/PENDING), og kategorisering.",
"curl_example": curl_example,
"result": result,
"error": error,
"is_demo": is_demo or (result == demo_data.MOCK_TRANSACTIONS),
"next_step": 6,
"prev_step": 4,
}))
# ---------------------------------------------------------------------------
# Step 6 — Events (v2)
# ---------------------------------------------------------------------------
@router.get("/demo/step/6", response_class=HTMLResponse)
async def step6(request: Request):
sess = _session(request)
s = get_settings()
is_demo = sess.get("demo_mode") or s.demo_mode
error = None
result_webhooks = None
webhook_registered = None
webhook_url = f"{s.app_base_url}/webhooks/tink"
curl_list = (
"# List registered webhooks (app-level token)\n"
"curl 'https://api.tink.com/api/v1/webhooks' \\\n"
" -H 'Authorization: Bearer $APP_TOKEN'"
)
curl_register = (
"# Register webhook endpoint\n"
"curl -X POST 'https://api.tink.com/api/v1/webhooks' \\\n"
" -H 'Authorization: Bearer $APP_TOKEN' \\\n"
" -H 'Content-Type: application/json' \\\n"
" -d '{\n"
' "url": "' + webhook_url + '",\n'
' "enabledEvents": [\n'
' "account-booked-transaction:created",\n'
' "account-pending-transaction:created"\n'
" ]\n"
" }'"
)
if is_demo:
result_webhooks = demo_data.MOCK_EVENTS_BOOKED
webhook_registered = {
"id": "wh-demo-001",
"url": webhook_url,
"enabledEvents": ["account-booked-transaction:created", "account-pending-transaction:created"],
"status": "ENABLED",
}
else:
try:
client = _client(log_cb=_logger(sess))
app_token_resp = await client.get_app_token(scope="user:create")
app_token = app_token_resp.get("access_token", "")
result_webhooks = await client.list_webhooks(app_token)
# Register our webhook if not already there
existing = [w for w in result_webhooks.get("webhooks", []) if w.get("url") == webhook_url]
if not existing:
webhook_registered = await client.register_webhook(app_token, webhook_url)
else:
webhook_registered = existing[0]
except Exception as e:
err_str = str(e)
if "404" in err_str:
result_webhooks = {"note": "Webhook API ikke tilgængeligt i sandbox — kun i produktion"}
webhook_registered = {
"url": webhook_url,
"enabledEvents": ["account-booked-transaction:created", "account-pending-transaction:created"],
"status": "ENABLED (eksempel)",
}
else:
error = err_str
return templates.TemplateResponse("step6.html", _ctx(request, {
"step": 6,
"title": "Webhooks & Events",
"subtitle": "Real-time Event Notifications",
"error": error,
"result_webhooks": result_webhooks,
"webhook_registered": webhook_registered,
"webhook_url": webhook_url,
"curl_list": curl_list,
"curl_register": curl_register,
"is_demo": is_demo,
"app_base_url": s.app_base_url,
"next_step": None,
"prev_step": 5,
}))
# ---------------------------------------------------------------------------
# API Request Log
# ---------------------------------------------------------------------------
@router.get("/demo/log", response_class=HTMLResponse)
async def api_log(request: Request):
sess = _session(request)
log = sess.get("api_log", [])
return templates.TemplateResponse("log.html", _ctx(request, {
"log": list(reversed(log)), # newest first
"log_count": len(log),
}))
@router.post("/demo/log/clear")
async def clear_log(request: Request):
sess = _session(request)
sess["api_log"] = []
return RedirectResponse("/demo/log", status_code=303)
# ---------------------------------------------------------------------------
# Webhook receiver
# ---------------------------------------------------------------------------
@router.post("/webhooks/tink")
async def webhook_receiver(request: Request):
"""Receive Tink webhook events (configure URL in Tink Console)."""
body = await request.json()
# In production you'd verify the signature and store events
print(f"[WEBHOOK] {json.dumps(body, indent=2)}")
return {"status": "received"}