How to Integrate FastAPI for banking with PostgreSQL for RAG

FastAPI for banking gives you a clean API layer for customer-facing and internal banking workflows. PostgreSQL gives you durable storage for transactions, customer profiles, and retrieval data for RAG. Put them together and you can build an AI agent that answers account questions, fetches policy or product context, and writes back structured events without turning your app into a pile of ad hoc scripts.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL 14+ instance
• •A FastAPI app with banking endpoints already defined
• •pip installed
• •These Python packages:
  • •fastapi
  • •uvicorn
  • •psycopg[binary] or psycopg2-binary
  • •sqlalchemy
  • •pydantic
• •A PostgreSQL database and user with permission to create tables
• •Environment variables configured:
  • •DATABASE_URL
  • •BANKING_API_BASE_URL
  • •BANKING_API_KEY

Integration Steps

1. •Set up the PostgreSQL connection layer.

For RAG, you want a schema that can store documents, embeddings, and metadata tied to banking entities like products, FAQs, or compliance notes.

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

DATABASE_URL = "postgresql+psycopg://rag_user:rag_pass@localhost:5432/banking_rag"

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

def init_db():
    with engine.begin() as conn:
        conn.execute(text("""
            CREATE TABLE IF NOT EXISTS rag_documents (
                id SERIAL PRIMARY KEY,
                source VARCHAR(100) NOT NULL,
                doc_type VARCHAR(50) NOT NULL,
                content TEXT NOT NULL,
                embedding VECTOR(1536),
                metadata JSONB,
                created_at TIMESTAMP DEFAULT NOW()
            );
        """))

If you are using pgvector, install the extension first:

from sqlalchemy import text

with engine.begin() as conn:
    conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector;"))

2. •Expose banking data through FastAPI endpoints.

Your agent needs a stable API surface to retrieve account context or transaction history before it performs retrieval over PostgreSQL.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Banking API")

class AccountResponse(BaseModel):
    account_id: str
    balance: float
    currency: str

@app.get("/accounts/{account_id}", response_model=AccountResponse)
def get_account(account_id: str):
    # Replace this with your real banking service lookup
    if account_id != "acc_123":
        raise HTTPException(status_code=404, detail="Account not found")

    return AccountResponse(
        account_id="acc_123",
        balance=12500.75,
        currency="USD"
    )

This is the contract your agent will call when it needs live account state before answering a question.

3. •Build the retrieval write path into PostgreSQL.

When new banking knowledge arrives — product docs, fee schedules, dispute workflows — persist it in Postgres so your agent can retrieve it later.

from sqlalchemy import text

def save_document(source: str, doc_type: str, content: str, metadata: dict):
    with SessionLocal() as session:
        session.execute(
            text("""
                INSERT INTO rag_documents (source, doc_type, content, metadata)
                VALUES (:source, :doc_type, :content, :metadata)
            """),
            {
                "source": source,
                "doc_type": doc_type,
                "content": content,
                "metadata": metadata,
            }
        )
        session.commit()

save_document(
    source="banking_policy",
    doc_type="faq",
    content="Wire transfers submitted after 4 PM are processed next business day.",
    metadata={"product": "wire_transfer", "region": "US"}
)

If you are generating embeddings externally, store them alongside the content. For example:

def save_embedding(document_id: int, embedding: list[float]):
    with SessionLocal() as session:
        session.execute(
            text("""
                UPDATE rag_documents
                SET embedding = :embedding
                WHERE id = :document_id
            """),
            {"embedding": embedding, "document_id": document_id}
        )
        session.commit()

4. •Add an agent service that calls FastAPI and queries PostgreSQL.

This is the core integration point. The agent gets live banking data from FastAPI and relevant context from Postgres in one flow.

import requests
from sqlalchemy import text

BANKING_API_BASE_URL = "http://localhost:8000"

def get_banking_context(account_id: str):
    resp = requests.get(f"{BANKING_API_BASE_URL}/accounts/{account_id}", timeout=5)
    resp.raise_for_status()
    return resp.json()

def retrieve_docs(query_term: str):
    with SessionLocal() as session:
        rows = session.execute(
            text("""
                SELECT id, source, doc_type, content
                FROM rag_documents
                WHERE content ILIKE :q OR metadata::text ILIKE :q
                ORDER BY created_at DESC
                LIMIT 5
            """),
            {"q": f"%{query_term}%"}
        ).fetchall()

        return [
            {
                "id": row.id,
                "source": row.source,
                "doc_type": row.doc_type,
                "content": row.content,
            }
            for row in rows
        ]

def build_agent_payload(account_id: str, query_term: str):
    account = get_banking_context(account_id)
    docs = retrieve_docs(query_term)

    return {
        "account": account,
        "retrieved_context": docs,
    }

That payload is what your LLM or rules engine uses to answer questions grounded in both live banking state and stored bank knowledge.

5. •Wire it into a single endpoint for downstream consumers.

This makes the integration usable by your internal tools or chat layer.

from fastapi import Depends

@app.get("/agent/context/{account_id}")
def agent_context(account_id: str):
    payload = build_agent_payload(account_id=account_id, query_term="wire transfer")
    return payload

Testing the Integration

Run both services first:

uvicorn main:app --reload --port 8000

Then verify the combined flow:

import requests

resp = requests.get("http://localhost:8000/agent/context/acc_123", timeout=5)
print(resp.status_code)
print(resp.json())

Expected output:

200
{
  "account": {
    "account_id": "acc_123",
    "balance": 12500.75,
    "currency": "USD"
  },
  "retrieved_context": [
    {
      "id": 1,
      "source": "banking_policy",
      "doc_type": "faq",
      "content": "Wire transfers submitted after 4 PM are processed next business day."
    }
  ]
}

If this returns both live account data and retrieved policy context, your RAG pipeline is wired correctly.

Real-World Use Cases

• •

  Customer support assistant

  • •Answer balance-related questions from FastAPI.
  • •Pull policy docs from PostgreSQL to explain fees, transfer cutoffs, or card replacement rules.
• •

  Compliance-aware agent

  • •Retrieve AML/KYC guidance from Postgres.
  • •Combine it with customer/account state from FastAPI before generating responses or escalation notes.
• •

  Internal banker copilot

  • •Summarize recent transactions from your banking API.
  • •Ground recommendations in product documentation stored in PostgreSQL for accurate next-best-action suggestions.

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.