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

Combining FastAPI for healthcare with PostgreSQL gives you a clean way to expose clinical workflows as APIs while keeping agent state, audit logs, and structured patient context in a durable database. In multi-agent systems, that matters because one agent can triage, another can summarize, and a third can persist decisions without losing traceability.

Prerequisites

• •Python 3.11+
• •FastAPI installed and running
• •uvicorn for local API serving
• •PostgreSQL 14+ running locally or in your environment
• •psycopg v3 or asyncpg for PostgreSQL access
• •A database user with permission to create tables and write rows
• •Optional: python-dotenv for environment variables
• •A healthcare domain model ready for:
  • •patient records
  • •encounter summaries
  • •agent messages / tool outputs

Integration Steps

1. •

   Set up your PostgreSQL connection settings

   Keep credentials out of code. For multi-agent systems, you want one shared persistence layer that every agent can read from and write to.

   import os
   from dotenv import load_dotenv
   
   load_dotenv()
   
   DB_HOST = os.getenv("DB_HOST", "localhost")
   DB_PORT = os.getenv("DB_PORT", "5432")
   DB_NAME = os.getenv("DB_NAME", "health_agents")
   DB_USER = os.getenv("DB_USER", "postgres")
   DB_PASSWORD = os.getenv("DB_PASSWORD", "postgres")
   
   DATABASE_URL = (
       f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
   )
   

2. •

   Create the database schema for agent state and healthcare records

   Use a small schema first. You need tables for patients, encounters, and agent events so each agent can coordinate through shared state.

   import psycopg
   
   schema_sql = """
   CREATE TABLE IF NOT EXISTS patients (
       id UUID PRIMARY KEY,
       full_name TEXT NOT NULL,
       date_of_birth DATE,
       created_at TIMESTAMPTZ DEFAULT NOW()
   );
   
   CREATE TABLE IF NOT EXISTS encounters (
       id UUID PRIMARY KEY,
       patient_id UUID REFERENCES patients(id),
       chief_complaint TEXT NOT NULL,
       summary TEXT,
       created_at TIMESTAMPTZ DEFAULT NOW()
   );
   
   CREATE TABLE IF NOT EXISTS agent_events (
       id UUID PRIMARY KEY,
       encounter_id UUID REFERENCES encounters(id),
       agent_name TEXT NOT NULL,
       event_type TEXT NOT NULL,
       payload JSONB NOT NULL,
       created_at TIMESTAMPTZ DEFAULT NOW()
   );
   """
   
   with psycopg.connect(DATABASE_URL) as conn:
       with conn.cursor() as cur:
           cur.execute(schema_sql)
       conn.commit()
   

3. •

   Build the FastAPI app and wire PostgreSQL into request handling

   FastAPI’s dependency injection is the right place to open database connections per request. For production, use a pool; for clarity here, this uses direct connections with psycopg.connect().

   from uuid import UUID, uuid4
   from fastapi import FastAPI, HTTPException
   from pydantic import BaseModel
   import psycopg
   
   app = FastAPI(title="Healthcare Multi-Agent API")
   
   class PatientIn(BaseModel):
       full_name: str
       date_of_birth: str | None = None
   
   class EncounterIn(BaseModel):
       patient_id: UUID
       chief_complaint: str
   
   @app.post("/patients")
   def create_patient(payload: PatientIn):
       patient_id = uuid4()
       with psycopg.connect(DATABASE_URL) as conn:
           with conn.cursor() as cur:
               cur.execute(
                   """
                   INSERT INTO patients (id, full_name, date_of_birth)
                   VALUES (%s, %s, %s)
                   """,
                   (patient_id, payload.full_name, payload.date_of_birth),
               )
           conn.commit()
       return {"id": str(patient_id), "full_name": payload.full_name}
   
   @app.post("/encounters")
   def create_encounter(payload: EncounterIn):
       encounter_id = uuid4()
       with psycopg.connect(DATABASE_URL) as conn:
           with conn.cursor() as cur:
               cur.execute(
                   "SELECT id FROM patients WHERE id = %s",
                   (payload.patient_id,),
               )
               if cur.fetchone() is None:
                   raise HTTPException(status_code=404, detail="Patient not found")
   
               cur.execute(
                   """
                   INSERT INTO encounters (id, patient_id, chief_complaint)
                   VALUES (%s, %s, %s)
                   """,
                   (encounter_id, payload.patient_id, payload.chief_complaint),
               )
           conn.commit()
       return {"id": str(encounter_id), "patient_id": str(payload.patient_id)}
   

4. •

   Add an endpoint for multi-agent event logging

   This is the part most teams skip. If you want multiple agents to collaborate safely in healthcare workflows, every tool call should be persisted with an event type and structured payload.

    from typing import Any
   
    class AgentEventIn(BaseModel):
        encounter_id: UUID
        agent_name: str
        event_type: str
        payload: dict[str, Any]
   
    @app.post("/agent-events")
    def log_agent_event(payload: AgentEventIn):
        event_id = uuid4()
        with psycopg.connect(DATABASE_URL) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO agent_events (id, encounter_id, agent_name, event_type, payload)
                    VALUES (%s, %s, %s, %s, %s::jsonb)
                    """,
                    (
                        event_id,
                        payload.encounter_id,
                        payload.agent_name,
                        payload.event_type,
                        json.dumps(payload.payload),
                    ),
                )
            conn.commit()
        return {"id": str(event_id), "status": "logged"}
   

5. •

   Expose a retrieval endpoint so agents can read shared context

   In a multi-agent system, one agent often needs the output of another. Make that explicit through API reads backed by PostgreSQL joins.

    @app.get("/encounters/{encounter_id}")
    def get_encounter(encounter_id: UUID):
        with psycopg.connect(DATABASE_URL) as conn:
            with conn.cursor(row_factory=psycopg.rows.dict_row) as cur:
                cur.execute(
                    """
                    SELECT e.id AS encounter_id,
                           e.chief_complaint,
                           e.summary,
                           p.full_name AS patient_name
                    FROM encounters e
                    JOIN patients p ON p.id = e.patient_id
                    WHERE e.id = %s
                    """,
                    (encounter_id,),
                )
                row = cur.fetchone()
   
                if row is None:
                    raise HTTPException(status_code=404, detail="Encounter not found")
   
                return row
   

Testing the Integration

Run the app:

uvicorn main:app --reload

Then verify it end-to-end:

import requests

base_url = "http://127.0.0.1:8000"

patient = requests.post(
    f"{base_url}/patients",
    json={"full_name": "Jane Doe", "date_of_birth": "1988-04-12"},
).json()

encounter = requests.post(
    f"{base_url}/encounters",
    json={
        "patient_id": patient["id"],
        "chief_complaint": "Shortness of breath",
    },
).json()

event = requests.post(
    f"{base_url}/agent-events",
    json={
        "encounter_id": encounter["id"],
        "agent_name": "triage-agent",
        "event_type": "assessment_created",
        "payload": {"risk_level": "high", "notes": "Escalate to clinician"},
    },
).json()

print(patient)
print(encounter)
print(event)

Expected output:

{'id': '...', 'full_name': 'Jane Doe'}
{'id': '...', 'patient_id': '...'}
{'id': '...', 'status': 'logged'}

Real-World Use Cases

• •

  Clinical intake orchestration

  • •One agent collects symptoms through FastAPI.
  • •Another stores structured intake data in PostgreSQL.
  • •A third generates escalation recommendations and logs them for audit.
• •

  Care coordination workflows

  • •Agents can read prior encounters from PostgreSQL.
  • •FastAPI exposes endpoints for scheduling updates, follow-ups, and task assignments.
  • •Every action gets persisted for traceability.
• •

  Compliance-friendly audit trails

  • •Store every tool call from each agent in agent_events.
  • •Query by patient ID or encounter ID during reviews.
  • •Keep operational visibility without burying logic inside prompts.

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.