How to Integrate FastAPI for banking with PostgreSQL for multi-agent systems

Combining FastAPI for banking with PostgreSQL gives you a clean way to expose banking workflows through an API while keeping agent state, transaction history, and audit trails in a durable relational store. For multi-agent systems, that matters because one agent can fetch account context, another can validate risk rules, and a third can persist decisions without stepping on each other.

Prerequisites

• •Python 3.11+
• •A running PostgreSQL instance
• •fastapi
• •uvicorn
• •psycopg2-binary or asyncpg
• •sqlalchemy
• •A FastAPI banking backend or service endpoint you can call from your app
• •Environment variables set for:
  • •DATABASE_URL
  • •BANKING_API_BASE_URL
  • •BANKING_API_KEY

Install the core packages:

pip install fastapi uvicorn sqlalchemy psycopg2-binary requests pydantic

Integration Steps

1. •Create the PostgreSQL connection layer

Use SQLAlchemy for connection pooling and predictable session handling. In multi-agent systems, this is the layer that stores shared state like task status, account snapshots, and agent decisions.

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+psycopg2://postgres:postgres@localhost:5432/agents")

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def init_db():
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS agent_tasks (
                id SERIAL PRIMARY KEY,
                task_id VARCHAR(64) UNIQUE NOT NULL,
                agent_name VARCHAR(100) NOT NULL,
                status VARCHAR(32) NOT NULL,
                payload JSONB NOT NULL,
                created_at TIMESTAMP DEFAULT NOW()
            )
        """))

2. •Wrap the FastAPI banking API in a service client

Your agents should not call the banking API directly from random places. Put it behind a client so retries, headers, and auth are centralized.

import os
import requests

class BankingClient:
    def __init__(self):
        self.base_url = os.getenv("BANKING_API_BASE_URL", "https://banking.example.com")
        self.api_key = os.getenv("BANKING_API_KEY", "")

    def _headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

    def get_account_balance(self, account_id: str):
        url = f"{self.base_url}/accounts/{account_id}/balance"
        resp = requests.get(url, headers=self._headers(), timeout=10)
        resp.raise_for_status()
        return resp.json()

    def create_transfer(self, source_account: str, destination_account: str, amount: float):
        url = f"{self.base_url}/transfers"
        payload = {
            "source_account": source_account,
            "destination_account": destination_account,
            "amount": amount,
        }
        resp = requests.post(url, json=payload, headers=self._headers(), timeout=10)
        resp.raise_for_status()
        return resp.json()

3. •Build the FastAPI app that coordinates both systems

This is where the integration becomes useful. The API receives an agent request, calls the banking service through its REST endpoints, then stores the result in PostgreSQL for traceability.

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from sqlalchemy import text
import uuid

app = FastAPI(title="Banking Agent Orchestrator")
banking_client = BankingClient()

class TransferRequest(BaseModel):
    task_id: str | None = None
    agent_name: str
    source_account: str
    destination_account: str
    amount: float

@app.on_event("startup")
def startup():
    init_db()

@app.post("/agent/transfers")
def orchestrate_transfer(req: TransferRequest, db: Session = Depends(lambda: next(get_db()))):
    task_id = req.task_id or str(uuid.uuid4())

    db.execute(
        text("""
            INSERT INTO agent_tasks (task_id, agent_name, status, payload)
            VALUES (:task_id, :agent_name, :status, :payload::jsonb)
            ON CONFLICT (task_id) DO UPDATE SET status = EXCLUDED.status
        """),
        {
            "task_id": task_id,
            "agent_name": req.agent_name,
            "status": "started",
            "payload": req.model_dump_json(),
        },
    )
    db.commit()

    transfer_result = banking_client.create_transfer(
        req.source_account,
        req.destination_account,
        req.amount,
    )

    db.execute(
        text("""
            UPDATE agent_tasks
            SET status = :status, payload = :payload::jsonb
            WHERE task_id = :task_id
        """),
        {
            "task_id": task_id,
            "status": "completed",
            "payload": {"request": req.model_dump(), "banking_result": transfer_result},
        },
    )
    db.commit()

    return {"task_id": task_id, "result": transfer_result}

4. •Add a read path for other agents

In multi-agent systems, one agent often needs to inspect what another agent already did. Store decisions in PostgreSQL and expose them through a small query endpoint.

from fastapi import HTTPException

@app.get("/agent/tasks/{task_id}")
def get_task(task_id: str):
    with engine.begin() as conn:
      row = conn.execute(
          text("SELECT task_id, agent_name, status, payload FROM agent_tasks WHERE task_id = :task_id"),
          {"task_id": task_id},
      ).mappings().first()

      if not row:
          raise HTTPException(status_code=404, detail="Task not found")

      return dict(row)

5. •Run the service and wire it into your agent workflow

Start the API with Uvicorn and point your agents at it as their coordination layer.

uvicorn main:app --reload --host 0.0.0.0 --port 8000

A typical flow looks like this:

• •Agent A checks account balance through the banking API client.
• •Agent B decides whether to initiate a transfer.
• •Agent C writes the decision and outcome to PostgreSQL.
• •Any downstream audit or reconciliation job reads from PostgreSQL.

Testing the Integration

Use FastAPI’s test client to verify both sides are working together. Mocking the banking service is fine for unit tests; in staging you should hit a real sandbox endpoint.

from fastapi.testclient import TestClient

client = TestClient(app)

def test_orchestrate_transfer():
    payload = {
        "agent_name": "risk-agent",
        "source_account": "ACC001",
        "destination_account": "ACC002",
        "amount": 125.50
    }

    response = client.post("/agent/transfers", json=payload)
    assert response.status_code == 200

    data = response.json()
    assert "task_id" in data
    assert "result" in data

print(client.get("/agent/tasks/some-task-id").status_code)

Expected output:

200 OK
{"task_id":"...","result":{"transfer_id":"trf_12345","status":"submitted"}}

Real-World Use Cases

• •

  Payment orchestration across agents

  • •One agent validates customer identity.
  • •Another checks limits and balances.
  • •A final agent submits transfers through the banking API and persists every step in PostgreSQL.
• •

  Fraud review pipelines

  • •Agents score transactions independently.
  • •Results are stored in PostgreSQL for explainability.
  • •The FastAPI layer exposes review actions to analysts or downstream services.
• •

  Loan servicing workflows

  • •One agent fetches account activity.
  • •Another computes delinquency risk.
  • •A third writes servicing recommendations to PostgreSQL and triggers bank-side actions through FastAPI endpoints.

If you want this production-ready next step is clear: add idempotency keys on every write path, use async database access for high concurrency, and separate read models from write models so your agents don’t fight over the same rows.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.