How to Build a Prior Authorization Packet Assembler with Arcade's Google Docs Toolkit

pip install arcadepy arcade-ai

export ARCADE_API_KEY="your_arcade_api_key"
export GOOGLE_CLIENT_ID="your_google_client_id"
export GOOGLE_CLIENT_SECRET="your_google_client_secret"

auth:
  providers:
    - id: google-docs-provider
      description: "Google OAuth for Docs access"
      enabled: true
      type: oauth2
      provider_id: google
      client_id: ${env:GOOGLE_CLIENT_ID}
      client_secret: ${env:GOOGLE_CLIENT_SECRET}

api:
  development: false
  host: 0.0.0.0
  port: 9099

tools:
  directors:
    - id: default
      enabled: true
      max_tools: 64

arcade engine start --config engine.yaml

import os
from arcadepy import Arcade
from datetime import datetime

class PriorAuthAssembler:
    def __init__(self):
        self.client = Arcade(api_key=os.environ.get("ARCADE_API_KEY"))
        self.authenticated_users = {}

    def get_client(self):
        return self.client

async def authenticate_user(self, user_id: str) -> dict:
    """Handle OAuth flow for healthcare staff member"""

    auth_response = await self.client.tools.authorize(
        tool_name="GoogleDocs.CreateBlankDocument",
        user_id=user_id
    )

    if auth_response.status != "completed":
        return {
            "requires_auth": True,
            "auth_url": auth_response.url,
            "message": "Complete Google authorization to access Docs"
        }

    await self.client.auth.wait_for_completion(auth_response)

    self.authenticated_users[user_id] = {
        "authenticated": True,
        "timestamp": datetime.now()
    }

    return {"authenticated": True}

async def ensure_authenticated(self, user_id: str) -> bool:
    """Verify user has valid Google Docs authorization"""

    if user_id not in self.authenticated_users:
        result = await self.authenticate_user(user_id)
        if result.get("requires_auth"):
            raise ValueError(f"User must authorize at: {result['auth_url']}")

    return True

async def create_document(self, user_id: str, title: str) -> dict:
    """Create new Google Docs document"""

    await self.ensure_authenticated(user_id)

    response = await self.client.tools.execute(
        tool_name="GoogleDocs.CreateBlankDocument",
        input={"title": title},
        user_id=user_id
    )

    return {
        "document_id": response.output.value["document_id"],
        "title": title,
        "url": response.output.value["url"]
    }

async def append_text(self, user_id: str, document_id: str, text: str) -> dict:
    """Insert text at end of document"""

    response = await self.client.tools.execute(
        tool_name="GoogleDocs.InsertText",
        input={
            "document_id": document_id,
            "text": text
        },
        user_id=user_id
    )

    return response.output.value

async def fetch_patient_data(self, patient_id: str) -> dict:
    """Retrieve patient information from EHR"""

    # Replace with actual EHR API integration
    return {
        "patient_name": "John Doe",
        "dob": "1985-03-15",
        "member_id": "MEM123456789",
        "insurance_provider": "Blue Cross Blue Shield",
        "policy_number": "BCBS-987654321",
        "pcp_name": "Dr. Sarah Johnson",
        "patient_phone": "(555) 123-4567",
        "patient_address": "123 Main St, Anytown, ST 12345"
    }

async def fetch_clinical_data(self, patient_id: str, encounter_id: str) -> dict:
    """Retrieve clinical documentation from EHR"""

    return {
        "primary_diagnosis": "Chronic Lower Back Pain",
        "icd10_code": "M54.5",
        "secondary_diagnoses": ["Sciatica (M54.4)", "Radiculopathy (M54.1)"],
        "procedure_name": "Lumbar MRI with Contrast",
        "cpt_code": "72148",
        "medical_necessity": "Patient presents with persistent lower back pain radiating to left leg for 8 weeks, unresponsive to conservative treatment.",
        "treatment_history": [
            {
                "name": "Physical Therapy",
                "date": "2025-08-01",
                "outcome": "Minimal improvement after 6 weeks"
            }
        ],
        "physician_notes": "Patient reports 7/10 pain severity. Physical exam reveals positive straight leg raise test. MRI recommended to rule out disc herniation."
    }

async def fetch_diagnostic_data(self, patient_id: str) -> dict:
    """Retrieve lab and imaging results"""

    return {
        "lab_results": [
            {
                "test_name": "Complete Blood Count",
                "date": "2025-09-10",
                "value": "14.2 g/dL",
                "reference_range": "13.5-17.5 g/dL",
                "status": "Normal"
            }
        ],
        "imaging_studies": [
            {
                "study_type": "Lumbar X-Ray",
                "date": "2025-08-20",
                "facility": "Anytown Radiology Center",
                "findings": "Mild degenerative changes at L4-L5. No acute fracture.",
                "impression": "Age-appropriate degenerative changes"
            }
        ]
    }

async def add_demographics_section(
    self,
    user_id: str,
    document_id: str,
    patient_data: dict
) -> None:
    """Add patient demographics to document"""

    text = f"""PATIENT DEMOGRAPHICS
====================

Patient Name: {patient_data['patient_name']}
Date of Birth: {patient_data['dob']}
Member ID: {patient_data['member_id']}
Insurance Provider: {patient_data['insurance_provider']}
Policy Number: {patient_data['policy_number']}
Primary Care Physician: {patient_data['pcp_name']}
Phone: {patient_data['patient_phone']}
Address: {patient_data['patient_address']}

"""

    await self.append_text(user_id, document_id, text)

async def add_clinical_section(
    self,
    user_id: str,
    document_id: str,
    clinical_data: dict
) -> None:
    """Add clinical documentation to document"""

    treatment_list = '\n'.join([
        f"- {t['name']} ({t['date']}): {t['outcome']}"
        for t in clinical_data['treatment_history']
    ])

    text = f"""CLINICAL DOCUMENTATION
======================

Primary Diagnosis: {clinical_data['primary_diagnosis']} (ICD-10: {clinical_data['icd10_code']})
Secondary Diagnoses: {', '.join(clinical_data['secondary_diagnoses'])}

Procedure Requested: {clinical_data['procedure_name']} (CPT: {clinical_data['cpt_code']})
Medical Necessity: {clinical_data['medical_necessity']}

Previous Treatments:
{treatment_list}

Clinical Notes:
{clinical_data['physician_notes']}

"""

    await self.append_text(user_id, document_id, text)

async def add_diagnostics_section(
    self,
    user_id: str,
    document_id: str,
    diagnostics: dict
) -> None:
    """Add lab and imaging results to document"""

    text = "DIAGNOSTIC RESULTS\n==================\n\nLab Results:\n"

    for lab in diagnostics['lab_results']:
        text += f"""
{lab['test_name']} ({lab['date']}):
  Result: {lab['value']}
  Reference: {lab['reference_range']}
  Status: {lab['status']}
"""

    text += "\nImaging Studies:\n"

    for imaging in diagnostics['imaging_studies']:
        text += f"""
{imaging['study_type']} ({imaging['date']}):
  Facility: {imaging['facility']}
  Findings: {imaging['findings']}
  Impression: {imaging['impression']}
"""

    await self.append_text(user_id, document_id, text)

async def add_metadata_section(self, user_id: str, document_id: str) -> None:
    """Add document generation metadata"""

    text = f"""

SUBMISSION INFORMATION
======================

Document Generated: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
Generated By: {user_id}
System: Prior Authorization Automated Assembler

Attestation: Information contained in this request is accurate and complete.

"""

    await self.append_text(user_id, document_id, text)

async def assemble_packet(
    self,
    user_id: str,
    patient_id: str,
    encounter_id: str
) -> dict:
    """Execute complete prior auth packet assembly"""

    try:
        await self.ensure_authenticated(user_id)

        # Retrieve all data
        patient_data = await self.fetch_patient_data(patient_id)
        clinical_data = await self.fetch_clinical_data(patient_id, encounter_id)
        diagnostics = await self.fetch_diagnostic_data(patient_id)

        # Create document
        doc_title = f"Prior Auth - {patient_data['patient_name']} - {clinical_data['cpt_code']}"
        document = await self.create_document(user_id, doc_title)

        # Assemble sections
        await self.add_demographics_section(user_id, document["document_id"], patient_data)
        await self.add_clinical_section(user_id, document["document_id"], clinical_data)
        await self.add_diagnostics_section(user_id, document["document_id"], diagnostics)
        await self.add_metadata_section(user_id, document["document_id"])

        return {
            "success": True,
            "document_id": document["document_id"],
            "document_url": document["url"],
            "title": doc_title,
            "timestamp": datetime.now().isoformat()
        }

    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }

class MultiUserAssembler:
    def __init__(self):
        self.assembler = PriorAuthAssembler()
        self.sessions = {}

    async def process_request(
        self,
        staff_email: str,
        patient_id: str,
        encounter_id: str
    ) -> dict:
        """Process request with staff-specific context"""

        result = await self.assembler.assemble_packet(
            user_id=staff_email,
            patient_id=patient_id,
            encounter_id=encounter_id
        )

        if result["success"]:
            self.sessions[staff_email] = {
                "last_document": result["document_id"],
                "last_assembly": result["timestamp"]
            }

        return result

class AssemblyError(Exception):
    """Base exception for assembly errors"""
    pass

class AuthenticationError(AssemblyError):
    def __init__(self, message: str, auth_url: str):
        self.message = message
        self.auth_url = auth_url
        super().__init__(message)

async def handle_error(error: Exception) -> dict:
    """Handle assembly errors"""

    if isinstance(error, AuthenticationError):
        return {
            "error_type": "authentication",
            "requires_action": True,
            "auth_url": error.auth_url
        }

    return {
        "error_type": "system",
        "requires_action": False,
        "message": str(error)
    }

import asyncio

async def execute_with_retry(func, max_retries: int = 3, *args, **kwargs):
    """Execute function with exponential backoff"""

    for attempt in range(max_retries):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = 2 ** attempt
            await asyncio.sleep(delay)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
multi_user_system = MultiUserAssembler()

class AssemblyRequest(BaseModel):
    patient_id: str
    encounter_id: str
    staff_email: str

class AssemblyResponse(BaseModel):
    status: str
    document_id: str | None = None
    document_url: str | None = None
    message: str | None = None

@app.post("/api/assemble", response_model=AssemblyResponse)
async def assemble_packet(request: AssemblyRequest):
    """Trigger prior auth packet assembly"""

    result = await multi_user_system.process_request(
        staff_email=request.staff_email,
        patient_id=request.patient_id,
        encounter_id=request.encounter_id
    )

    if not result["success"]:
        raise HTTPException(status_code=500, detail=result.get("error"))

    return AssemblyResponse(
        status="completed",
        document_id=result["document_id"],
        document_url=result["document_url"],
        message="Prior authorization packet assembled"
    )

@app.get("/api/status/{staff_email}")
async def check_status(staff_email: str):
    """Check authentication status"""

    session = multi_user_system.sessions.get(staff_email)

    return {
        "authenticated": session is not None,
        "last_assembly": session.get("last_assembly") if session else None
    }

import asyncio

async def fetch_all_data(self, patient_id: str, encounter_id: str) -> tuple:
    """Fetch data from multiple sources concurrently"""

    results = await asyncio.gather(
        self.fetch_patient_data(patient_id),
        self.fetch_clinical_data(patient_id, encounter_id),
        self.fetch_diagnostic_data(patient_id)
    )

    return results

from collections import OrderedDict
import time

class ToolCache:
    def __init__(self, max_size: int = 100, ttl: int = 3600):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl

    def set(self, user_id: str, tools: list):
        if user_id in self.cache:
            self.cache.move_to_end(user_id)

        self.cache[user_id] = {
            "tools": tools,
            "timestamp": time.time()
        }

        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)

    def get(self, user_id: str):
        entry = self.cache.get(user_id)
        if not entry:
            return None

        if time.time() - entry["timestamp"] > self.ttl:
            del self.cache[user_id]
            return None

        self.cache.move_to_end(user_id)
        return entry["tools"]

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {
            "completed": 0,
            "failed": 0,
            "avg_time": 0.0
        }
        self.times = []

    async def track(self, user_id: str, duration: float, success: bool):
        """Track assembly performance"""

        self.times.append(duration)

        if success:
            self.metrics["completed"] += 1
        else:
            self.metrics["failed"] += 1

        self.metrics["avg_time"] = sum(self.times) / len(self.times)

    def report(self) -> dict:
        """Generate health report"""

        total = self.metrics["completed"] + self.metrics["failed"]
        success_rate = (self.metrics["completed"] / total * 100) if total > 0 else 0

        return {
            "status": "healthy" if success_rate > 95 else "degraded",
            "total": total,
            "success_rate": f"{success_rate:.2f}%",
            "avg_time": f"{self.metrics['avg_time']:.2f}s"
        }

async def submit_via_email(
    self,
    user_id: str,
    document_url: str,
    insurer_email: str,
    patient_name: str
):
    """Email packet to insurance provider"""

    await self.client.tools.execute(
        tool_name="Gmail.SendEmail",
        input={
            "to": insurer_email,
            "subject": f"Prior Authorization Request - {patient_name}",
            "body": f"Prior authorization packet: {document_url}"
        },
        user_id=user_id
    )

async def schedule_followup(
    self,
    user_id: str,
    patient_name: str,
    days: int = 3
):
    """Create calendar reminder"""

    from datetime import timedelta
    followup_date = datetime.now() + timedelta(days=days)

    await self.client.tools.execute(
        tool_name="GoogleCalendar.CreateEvent",
        input={
            "summary": f"Follow up - {patient_name} prior auth",
            "start_time": followup_date.isoformat(),
            "duration_minutes": 15
        },
        user_id=user_id
    )

async def notify_team(
    self,
    document_url: str,
    patient_name: str,
    channel: str = "#prior-auth"
):
    """Send Slack notification"""

    await self.client.tools.execute(
        tool_name="Slack.SendMessage",
        input={
            "channel": channel,
            "text": f"Prior auth completed for {patient_name}: {document_url}"
        },
        user_id="system_user"
    )

from celery import Celery

app = Celery('prior_auth', broker='redis://localhost:6379')

@app.task
async def assemble_task(user_id: str, patient_id: str, encounter_id: str):
    """Background assembly task"""
    assembler = PriorAuthAssembler()
    return await assembler.assemble_packet(user_id, patient_id, encounter_id)

@app.get("/health")
async def health_check():
    """System health endpoint"""

    monitor = PerformanceMonitor()
    report = monitor.report()

    return {
        "status": report["status"],
        "metrics": report
    }