Building a New Patient Intake Agent: Forms, Medical History, and Pre-Visit Coordination

Why Intake Is the Perfect Use Case for AI Agents

New patient intake is simultaneously critical and frustrating. Patients fill out pages of paperwork in the waiting room, staff manually enter the data, and errors propagate into the medical record. An AI intake agent digitizes this entire workflow: it collects information conversationally, validates entries in real time, and pushes structured data directly to the EMR.

The result is a faster, more accurate process that reduces the average intake time from 20 minutes of paperwork to a 5-minute guided conversation.

Form Schema Definition

Define the intake form as a structured schema. This lets the agent know what information to collect and how to validate each field.

from dataclasses import dataclass, field
from typing import Optional, Any
from enum import Enum
from datetime import date


class FieldType(Enum):
    TEXT = "text"
    DATE = "date"
    BOOLEAN = "boolean"
    SELECT = "select"
    MULTI_SELECT = "multi_select"
    PHONE = "phone"
    EMAIL = "email"


@dataclass
class FormField:
    name: str
    label: str
    field_type: FieldType
    required: bool = True
    options: list[str] = field(default_factory=list)
    validation_regex: Optional[str] = None
    help_text: str = ""


INTAKE_FORM = [
    FormField("first_name", "First Name", FieldType.TEXT),
    FormField("last_name", "Last Name", FieldType.TEXT),
    FormField("dob", "Date of Birth", FieldType.DATE),
    FormField(
        "phone", "Phone Number", FieldType.PHONE,
        validation_regex=r"^\+?1?\d{10}$",
    ),
    FormField("email", "Email Address", FieldType.EMAIL),
    FormField(
        "gender", "Gender", FieldType.SELECT,
        options=["Male", "Female", "Non-binary",
                 "Prefer not to say"],
    ),
    FormField(
        "allergies", "Known Allergies",
        FieldType.MULTI_SELECT, required=False,
        options=["Penicillin", "Latex", "Lidocaine",
                 "Aspirin", "Ibuprofen", "None"],
    ),
    FormField(
        "medications", "Current Medications",
        FieldType.TEXT, required=False,
        help_text="List all current medications and dosages",
    ),
    FormField(
        "conditions", "Medical Conditions",
        FieldType.MULTI_SELECT, required=False,
        options=["Diabetes", "Heart Disease", "Hypertension",
                 "Asthma", "Bleeding Disorder",
                 "Joint Replacement", "None"],
    ),
    FormField(
        "emergency_name", "Emergency Contact Name",
        FieldType.TEXT,
    ),
    FormField(
        "emergency_phone", "Emergency Contact Phone",
        FieldType.PHONE,
        validation_regex=r"^\+?1?\d{10}$",
    ),
]

Data Validation Engine

Raw patient input needs validation before it enters the medical record. The validation engine checks formats, flags medically relevant combinations, and asks follow-up questions when needed.

import re
from datetime import datetime


class ValidationResult:
    def __init__(self, valid: bool, error: str = "",
                 warnings: list[str] = None):
        self.valid = valid
        self.error = error
        self.warnings = warnings or []


class IntakeValidator:
    def validate_field(
        self, field_def: FormField, value: Any,
    ) -> ValidationResult:
        if field_def.required and not value:
            return ValidationResult(
                False,
                f"{field_def.label} is required.",
            )

        if not value:
            return ValidationResult(True)

        if field_def.field_type == FieldType.DATE:
            return self._validate_date(value, field_def.label)
        elif field_def.field_type == FieldType.PHONE:
            return self._validate_phone(value)
        elif field_def.field_type == FieldType.EMAIL:
            return self._validate_email(value)
        elif field_def.field_type == FieldType.SELECT:
            if value not in field_def.options:
                return ValidationResult(
                    False,
                    f"Please select from: "
                    f"{', '.join(field_def.options)}",
                )
        elif field_def.validation_regex:
            if not re.match(field_def.validation_regex, value):
                return ValidationResult(
                    False, f"Invalid format for {field_def.label}."
                )

        return ValidationResult(True)

    def _validate_date(self, value, label):
        try:
            parsed = datetime.strptime(value, "%Y-%m-%d").date()
            if parsed > date.today():
                return ValidationResult(
                    False, f"{label} cannot be in the future."
                )
            if parsed.year < 1900:
                return ValidationResult(
                    False, f"{label} year seems incorrect."
                )
            return ValidationResult(True)
        except ValueError:
            return ValidationResult(
                False,
                f"Please enter {label} as YYYY-MM-DD.",
            )

    def _validate_phone(self, value):
        digits = re.sub(r"\D", "", value)
        if len(digits) < 10 or len(digits) > 11:
            return ValidationResult(
                False, "Phone number must be 10 digits."
            )
        return ValidationResult(True)

    def _validate_email(self, value):
        pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z]{2,}$"
        if not re.match(pattern, value):
            return ValidationResult(
                False, "Please enter a valid email address."
            )
        return ValidationResult(True)

    def check_medical_alerts(
        self, intake_data: dict,
    ) -> list[str]:
        alerts = []
        conditions = intake_data.get("conditions", [])
        allergies = intake_data.get("allergies", [])
        medications = intake_data.get("medications", "")

        if "Bleeding Disorder" in conditions:
            alerts.append(
                "ALERT: Patient reports bleeding disorder. "
                "Verify coagulation status before procedures."
            )
        if "Latex" in allergies:
            alerts.append(
                "ALERT: Latex allergy. Use nitrile gloves."
            )
        if "blood thinner" in medications.lower():
            alerts.append(
                "ALERT: Patient on blood thinners. "
                "Consult with physician before extractions."
            )
        return alerts

EMR Integration Layer

Once validated, the intake data needs to flow into the practice's electronic medical record system. This adapter layer handles the translation between the agent's data format and the EMR's API.

from typing import Protocol


class EMRAdapter(Protocol):
    async def create_patient(self, data: dict) -> str: ...
    async def update_medical_history(
        self, patient_id: str, history: dict,
    ) -> bool: ...


class OpenDentalAdapter:
    def __init__(self, api_base: str, api_key: str):
        self.api_base = api_base
        self.headers = {"Authorization": f"ODFHIR {api_key}"}

    async def create_patient(self, data: dict) -> str:
        import httpx
        payload = {
            "LName": data["last_name"],
            "FName": data["first_name"],
            "Birthdate": data["dob"],
            "HmPhone": data.get("phone", ""),
            "Email": data.get("email", ""),
            "Gender": self._map_gender(data.get("gender")),
        }
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self.api_base}/patients",
                json=payload,
                headers=self.headers,
            )
            resp.raise_for_status()
            return resp.json()["PatNum"]

    async def update_medical_history(
        self, patient_id: str, history: dict,
    ) -> bool:
        import httpx
        allergies = history.get("allergies", [])
        conditions = history.get("conditions", [])
        async with httpx.AsyncClient() as client:
            for allergy in allergies:
                await client.post(
                    f"{self.api_base}/allergies",
                    json={
                        "PatNum": patient_id,
                        "DefNum": self._allergy_code(allergy),
                        "StatusIsActive": True,
                    },
                    headers=self.headers,
                )
            for condition in conditions:
                await client.post(
                    f"{self.api_base}/diseases",
                    json={
                        "PatNum": patient_id,
                        "DiseaseDefNum": self._condition_code(
                            condition
                        ),
                    },
                    headers=self.headers,
                )
        return True

    def _map_gender(self, gender: str) -> int:
        return {"Male": 0, "Female": 1}.get(gender, 2)

Document Collection Coordinator

The agent tracks required documents — insurance cards, photo ID, referral letters — and sends reminders for missing items.

@dataclass
class RequiredDocument:
    doc_type: str
    description: str
    is_uploaded: bool = False
    upload_url: Optional[str] = None


class DocumentCollector:
    REQUIRED_DOCS = [
        RequiredDocument("insurance_front", "Insurance card (front)"),
        RequiredDocument("insurance_back", "Insurance card (back)"),
        RequiredDocument("photo_id", "Photo ID (driver license or passport)"),
    ]

    async def get_missing_documents(
        self, patient_id: str, db,
    ) -> list[RequiredDocument]:
        uploaded = await db.fetch("""
            SELECT doc_type FROM patient_documents
            WHERE patient_id = $1
        """, patient_id)
        uploaded_types = {r["doc_type"] for r in uploaded}

        return [
            doc for doc in self.REQUIRED_DOCS
            if doc.doc_type not in uploaded_types
        ]

    async def send_upload_reminders(
        self, patient_id: str, missing: list[RequiredDocument],
        sms_client, phone: str,
    ):
        if not missing:
            return
        doc_list = ", ".join(d.description for d in missing)
        await sms_client.send(phone, (
            f"Before your visit, please upload: {doc_list}. "
            f"Use this link: https://intake.example.com/"
            f"upload/{patient_id}"
        ))

FAQ

How does the agent handle patients who are not comfortable entering medical information digitally?

The agent supports a hybrid mode where the front desk staff can complete the intake form on the patient's behalf during a phone call. The conversational interface works the same way — the staff member reads the questions and enters responses. The system also supports a paper-to-digital workflow where scanned forms are processed via OCR.

What happens if the EMR system is temporarily unavailable?

The intake agent stores the validated data locally in a staging table and marks it for sync. A background job retries the EMR push on an exponential backoff schedule. The patient record is created in the EMR as soon as connectivity is restored, and staff receive an alert if any records remain unsynced for more than four hours.

How is patient data protected during the intake process?

All data is encrypted in transit using TLS and at rest using AES-256 encryption. The agent does not store raw medical data in conversation logs — only field identifiers and validation results are logged. The system implements role-based access controls, and all data handling complies with HIPAA requirements including audit logging of every access event.

---

#PatientIntake #HealthcareAI #EMRIntegration #DigitalForms #Python #AgenticAI #LearnAI #AIEngineering