How to Integrate FastAPI for healthcare with PostgreSQL for startups

FastAPI for healthcare gives you a clean way to expose clinical workflows, triage endpoints, and agent actions over HTTP. PostgreSQL gives you durable storage for patient events, audit logs, model outputs, and queue state. Put them together and you get the backbone for a startup-grade AI agent system that can accept requests, persist context, and serve regulated workflows without losing state.

Prerequisites

• •Python 3.10+
• •A running PostgreSQL instance
• •fastapi
• •uvicorn
• •psycopg2-binary or asyncpg
• •sqlalchemy
• •A FastAPI app configured for your healthcare workflow
• •Database credentials with permission to create tables
• •Network access from your app to PostgreSQL

Install the core packages:

pip install fastapi uvicorn sqlalchemy psycopg2-binary pydantic

Integration Steps

1. •Create the FastAPI healthcare app

Start with a small API that accepts a clinical intake payload. In production, this is where your agent would receive symptoms, claims data, prior auth requests, or care-routing tasks.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Healthcare Agent API")

class PatientIntake(BaseModel):
    patient_id: str
    age: int
    symptoms: list[str]
    urgency: str

@app.post("/intake")
async def create_intake(payload: PatientIntake):
    return {"status": "received", "patient_id": payload.patient_id}

2. •Configure the PostgreSQL connection

Use SQLAlchemy so your app can scale beyond one-off queries. For startups, this keeps your DB layer maintainable when you add migrations, retries, and audit tables.

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

DATABASE_URL = "postgresql+psycopg2://health_user:health_pass@localhost:5432/healthdb"

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

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

3. •Create the table for intake records

Store every request before you run any agent logic. In healthcare workflows, persistence first is the safer pattern because it gives you traceability and recovery if downstream steps fail.

from sqlalchemy import Table, Column, Integer, String, MetaData, JSON

metadata = MetaData()

intake_records = Table(
    "intake_records",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("patient_id", String(64), nullable=False),
    Column("age", Integer, nullable=False),
    Column("symptoms", JSON, nullable=False),
    Column("urgency", String(32), nullable=False),
)

metadata.create_all(engine)

4. •Write the FastAPI endpoint that persists to PostgreSQL

This is the actual integration point. The endpoint receives data from the healthcare workflow and writes it into PostgreSQL using a parameterized insert.

from fastapi import Depends
from sqlalchemy.orm import Session

@app.post("/intake/store")
async def store_intake(payload: PatientIntake, db: Session = Depends(get_db)):
    stmt = text("""
        INSERT INTO intake_records (patient_id, age, symptoms, urgency)
        VALUES (:patient_id, :age, :symptoms::jsonb, :urgency)
        RETURNING id
    """)

    result = db.execute(stmt, {
        "patient_id": payload.patient_id,
        "age": payload.age,
        "symptoms": payload.symptoms.__str__().replace("'", '"'),
        "urgency": payload.urgency,
    })
    record_id = result.scalar()
    db.commit()

    return {"status": "stored", "record_id": record_id}

5. •Add a read endpoint for agent retrieval

Your AI agent will need context on follow-up calls. Querying PostgreSQL lets you fetch prior intake history and use it in routing or summarization.

@app.get("/intake/{patient_id}")
async def get_intake(patient_id: str):
    with engine.connect() as conn:
        rows = conn.execute(
            text("SELECT id, patient_id, age, symptoms, urgency FROM intake_records WHERE patient_id = :patient_id"),
            {"patient_id": patient_id},
        ).mappings().all()

    return {"records": [dict(row) for row in rows]}

Testing the Integration

Run the app:

uvicorn main:app --reload

Then test with a request:

import requests

payload = {
    "patient_id": "P12345",
    "age": 42,
    "symptoms": ["fever", "cough"],
    "urgency": "high"
}

r = requests.post("http://127.0.0.1:8000/intake/store", json=payload)
print(r.status_code)
print(r.json())

r2 = requests.get("http://127.0.0.1:8000/intake/P12345")
print(r2.json())

Expected output:

200
{'status': 'stored', 'record_id': 1}
{'records': [{'id': 1, 'patient_id': 'P12345', 'age': 42, 'symptoms': ['fever', 'cough'], 'urgency': 'high'}]}

Real-World Use Cases

• •Clinical triage agents that capture symptoms through FastAPI and persist routing decisions in PostgreSQL for later review.
• •Prior authorization workflows where an agent collects documentation status and stores each step as an auditable record.
• •Care coordination systems that maintain patient interaction history so downstream agents can summarize context before making recommendations.

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.