earl-sdk 0.7.0

Python SDK for Earl Medical Evaluation Platform

Earl SDK for Python

Python SDK for the Earl Medical Evaluation Platform. Evaluate your medical AI/doctor chatbots against realistic patient simulations.

What's New

• Token cache keyed on resolved org_id - The on-disk token cache is now keyed on the org_id claim returned by Auth0 rather than the organization argument you passed in. This fixes collisions when the same client resolves different orgs and invalidates caches created by older SDK versions the first time they're read — you'll be asked to re-login once.
• Interactive Terminal UI - Rich terminal interface for exploring the platform, chatting with patients, running simulations, and comparing results -- all from your terminal
• 🔐 Client-Driven Mode - Run evaluations when your doctor API is behind a VPN or firewall. You control the conversation loop from your own infrastructure.
• Pipelines - Evaluation configurations are now called "pipelines" (previously "profiles")
• Flexible Authentication - External doctor APIs support both X-API-Key and Authorization: Bearer headers

Installation

pip install earl-sdk

With the interactive UI (adds rich and questionary):

pip install "earl-sdk[ui]"

With OS-keyring-backed secret storage (recommended; macOS Keychain, Windows Credential Vault, Linux Secret Service):

pip install "earl-sdk[secure]"
# or combine:
pip install "earl-sdk[ui,secure]"

Or install from source:

cd sdk
pip install -e ".[ui,secure]"

Quick Start

from earl_sdk import EarlClient, DoctorApiConfig

# Initialize with your Auth0 M2M credentials
client = EarlClient(
    client_id="your-m2m-client-id",
    client_secret="your-m2m-client-secret",
    organization="org_xxx",  # Your Auth0 organization ID
    environment="test",      # "test" or "prod" (default)
)

# Browse available evaluation cases
cases = client.cases.list()
for c in cases:
    print(f"  {c['case_id']}: {c['name']}")

# Optional: platform-wide verifier catalog (GET /additional-verifiers) — generic gates + scoring dims
catalog = client.verifiers.list()

# Create a pipeline from a pre-defined case
# The case includes: patient, case-specific verifiers, default hard gates + scoring dims
pipeline = client.pipelines.create(
    name="my-evaluation",
    case_id="carla-hypertension-yasmin",
    doctor_config=DoctorApiConfig.external(
        api_url="https://your-doctor-api.com/chat",
        api_key="your-api-key",
    ),
    max_turns=10,
)

# Run a simulation
simulation = client.simulations.create(
    pipeline_name=pipeline.name,
    num_episodes=1,
)

# Wait for completion
completed = client.simulations.wait_for_completion(simulation.id)

# Get the report — structured into hard gates, scoring dimensions, case verifiers
report = client.simulations.get_report(simulation.id)
for ep in report["episodes"]:
    for g in ep.get("hard_gates", []):
        print(f"  [{'PASS' if g['passed'] else 'FAIL'}] {g['id']}")
    for d in ep.get("scoring_dimensions", []):
        if d.get("activated"):
            print(f"  {d['score']}/4  {d['id']}")
    for v in ep.get("case_verifiers", []):
        if v.get("triggered"):
            print(f"  {v['points_awarded']:+d}pts  {v['id']}")

Generic verifier catalog (API)

Clinical cases bundle scenario-specific case verifiers; the platform also publishes a generic Lumos catalog (hard gates + scoring dimensions) at GET /api/v1/additional-verifiers. Use client.verifiers.list() in code. The interactive CLI loads this catalog when you add optional extra verifiers (not the union of per-case defaults scraped from cases).

sequenceDiagram
    participant UI as earl interactive
    participant SDK as EarlClient.verifiers
    participant API as Earl HTTP API
    UI->>SDK: list()
    SDK->>API: GET /additional-verifiers
    API-->>SDK: catalog JSON
    SDK-->>UI: parsed paths for multi-select

Override URL if needed: EARL_VERIFIERS_API_URL or service_api_urls["verifiers"] (base must end with /api/v1).

Interactive Terminal UI

The SDK includes a rich interactive terminal UI for exploring the platform without writing any code. Install with the ui extra and launch:

pip install "earl-sdk[ui]"
earl-ui

Or run as a module:

python -m earl_sdk.interactive

Features

Feature	Description
Chat with Patient	Be the doctor in a live conversation with a simulated patient; when the orchestrator stores patient API insights on each turn (metadata.insights), the CLI shows a short Patient insights panel under that reply (trust, mood, thoughts, etc.)
Run Simulation	Evaluate a doctor API against simulated patients and get scored results
Browse Simulations	Inspect past runs: episodes, dialogues, judge scores, and full reports
Compare Runs	Side-by-side delta view of 2-5 simulations across all dimensions
Explore Catalog	Browse available dimensions, patients, and pipelines on the platform
Configuration	Manage auth credentials, doctor API endpoints, and preferences

Local Storage

The UI stores data locally in ~/.earl/:

Path	Contents
~/.earl/config.json	Authentication profiles, doctor configurations, preferences
~/.earl/runs/	Simulation run metadata and reports for offline comparison

Credentials are stored in the OS keyring when earl-sdk[secure] is installed (recommended). Without the secure extra, secrets fall back to base64 obfuscation inside ~/.earl/config.json — which is not encryption. Check and migrate with:

earl auth backend                 # shows the active backend
earl auth migrate-secrets         # moves any base64 secrets into the keyring

Auth0 access tokens are cached on disk under ~/.earl/tokens/ (mode 0600) so long-running agent workflows do not re-authenticate on every call. Set EARL_NO_TOKEN_CACHE=1 to disable. For production CI use, prefer environment variables (EARL_CLIENT_ID, EARL_CLIENT_SECRET, EARL_ORG_ID).

First-Time Setup

On first launch, the UI will guide you through adding an authentication profile:

1. Choose Configuration > Auth Profiles > Add Profile
2. Enter your Auth0 M2M client_id, client_secret, and organization ID
3. Select an environment (local, test, or prod)
4. The UI tests the connection and saves the profile locally

Once configured, all other features become available.

Command-Line Interface (earl)

The SDK also ships an explicit command-based CLI for scripting and automation.

# install from source
cd sdk
pip install -e ".[ui]"

# show command tree
earl --help

Common commands

# Auth profile (prompts for secret; never put it on the command line)
earl auth profile add --name test --client-id "$EARL_CLIENT_ID" --env test
earl auth profile use test
earl auth test

# Browse catalog
earl cases list
earl patients list --limit 20

# Pipelines and simulations
earl pipelines list
earl simulations start --pipeline my-pipeline --num-episodes 3 --parallel-count 2
earl simulations wait <simulation_id>
earl simulations report <simulation_id> --save report.json

# Client-driven doctor loop helpers
earl simulations pending --simulation-id <simulation_id>
earl simulations respond <simulation_id> <episode_id> --message "Thanks for sharing that."

# Interactive chat flow (reuses existing ui implementation)
earl chat start

Scripting & LLM-agent features

The CLI is designed to be equally usable by humans and by LLM-driven agents:

• earl schema emits the entire command tree in a machine-readable form so agents can discover every flag without scraping --help output:

  earl schema --format json | jq '.commands | keys'
  earl schema --format markdown > earl-cli.md   # docs-site friendly
  

• --json is a shortcut for --output json, suitable for jq pipelines.
• --dry-run on mutating commands (pipelines create/update/delete, simulations start/stop/respond, dimensions create, auth profile add/delete, doctor add/delete) prints the resolved payload and exits without contacting the API and without requiring credentials. Secrets in the payload are redacted. Great for previewing automation before it runs:

  earl --dry-run pipelines create --name eval --case-id my-case --doctor internal
  

• --debug enables structured request logging on stderr (stdout stays a clean JSON stream).
• --quiet suppresses non-essential success messages.

Shell tab completion (optional):

pip install 'earl-sdk[complete]'
eval "$(register-python-argcomplete earl)"     # bash / zsh

Device Flow setup (interactive browser login)

For humans and for agents running on a developer workstation, prefer the OAuth 2.0 Device Authorization Grant over M2M client secrets. One-time setup per env, then earl auth login handles the rest.

In the Auth0 dashboard (once per environment), create a new Native application and copy its Client ID:

1. Applications → Create Application → Name: EARL CLI (<env>) → Type: Native.

2. Settings → Advanced Settings → Grant Types: enable Device Code and Refresh Token; leave the rest disabled.

3. Settings → Advanced → OAuth → Token Endpoint Authentication Method: None (it's a public client).

4. Settings → Organizations:

   • Type of Users: Business Users
   • Login Flow: No Prompt (Auth0's Device Flow does not render the in-browser org picker — the CLI handles that for you; see below).

5. APIs → your EARL API (matching audience) → Settings → Allow Offline Access: ON (required for Auth0 to issue refresh tokens).

6. (Optional, recommended for nicer UX) Actions → Library → Build Custom: create a Post-Login action that adds namespaced claims so the CLI can print a friendly org name + user email. Deploy and attach it to the Login flow:

   exports.onExecutePostLogin = async (event, api) => {
     if (event.organization?.id) {
       api.accessToken.setCustomClaim("https://earl/org_id", event.organization.id);
       api.accessToken.setCustomClaim("https://earl/org_name", event.organization.display_name ?? event.organization.name);
     }
     if (event.user?.email) {
       api.accessToken.setCustomClaim("https://earl/email", event.user.email);
     }
   };
   

   (Auth0 also surfaces org_id / org_name as top-level claims when Organizations is enabled on the app — the CLI reads both forms.)

7. Copy the Client ID — it's public; you can safely commit it or share it.

On your laptop:

# Register the public client_id once per env. Share this command with your
# team — the client_id itself is not a secret.
earl auth device-clients set --env dev <native-app-client-id>

# Sign in. Opens your browser. After Auth0 authenticates you, the CLI calls
# the orchestrator's `GET /api/v1/auth/my-orgs` endpoint to list every
# organization you belong to on this tenant, and then:
#   • auto-selects if you only belong to one, or
#   • shows a numbered picker in the terminal if you belong to several.
# No second browser trip is needed — the CLI exchanges the refresh token for
# an org-scoped access token in the background. The refresh token lives in
# the OS keyring (fallback: base64 in ~/.earl/config.json).
earl auth login --env dev

# Inspect which orgs the current profile can pick from:
earl auth my-orgs
earl --json auth my-orgs   # machine-readable

# Verify.
earl --json auth test

# If you belong to multiple orgs and want more than one profile, sign in
# multiple times:
earl auth login --env dev   # pick Org A in the CLI prompt → profile 'device-dev-org-a'
earl auth login --env dev   # pick Org B in the CLI prompt → profile 'device-dev-org-b'
# then switch any time with:
earl auth profile activate device-dev-org-b

# (Optional) remove a local session. Does NOT revoke the Auth0 grant;
# visit the Auth0 "Authorized Applications" page to do that.
earl auth logout --name device-dev-org-a

Device-flow profiles carry auth_kind=device. When the cached access token expires (~1 hour), the SDK silently trades the refresh token for a new one via Auth0's /oauth/token. Only when that refresh call fails (revoked/rotated/user removed from org) does the CLI prompt you to re-run earl auth login.

Advanced: power users (CI, scripted setups, first-login-on-fresh-tenant) can bypass the picker entirely by pre-declaring the org:

earl auth login --env dev --organization org_abc123

When --organization is passed the CLI sends organization=<id> straight to Auth0 on the device-authorization request and skips the /auth/my-orgs round-trip.

How the multi-org picker works

  ┌──────────┐   device auth (no org) ┌──────┐     refresh(org=X)     ┌──────┐
  │  CLI     │ ─────────────────────► │Auth0 │ ─────────────────────► │Auth0 │
  │          │ ◄───── access_token ── │      │ ◄── org-scoped AT ──── │      │
  │          │        (+ refresh)     └──────┘                        └──────┘
  │          │
  │          │   GET /auth/my-orgs (Bearer AT)   ┌────────────────┐  Mgmt API
  │          │ ──────────────────────────────► │ Orchestrator   │ ──────────►
  │          │ ◄── organizations: [...] ────── │                │ ◄── orgs ──
  └──────────┘                                  └────────────────┘

The no-org access token is only accepted by the single orchestrator endpoint GET /api/v1/auth/my-orgs (enforced by a dedicated FastAPI dependency, authenticate_request_org_discovery, with a CI guard to prevent it from being reused elsewhere). Every other endpoint requires a token whose org_id claim matches the requested resource, exactly as before — selecting an org is mandatory, not optional.

Man page

An installable man page is included at sdk/man/earl.1.

Environments

Earl provides three common environments:

Environment	Description	API URL
local	Local orchestrator	http://localhost:8006
dev	Development	https://earl-api.thelumos.dev
test	Staging / QA	https://earl-api.thelumos.xyz
prod	Production (default)	https://earl-api.thelumos.ai

from earl_sdk import EarlClient, Environment

# Test environment
test_client = EarlClient(
    client_id="test-client-id",
    client_secret="test-secret",
    organization="org_xxx",
    environment="test",
)

# Production environment (default)
prod_client = EarlClient(
    client_id="prod-client-id",
    client_secret="prod-secret",
    organization="org_xxx",
)

# Check which environment you're connected to
print(f"Environment: {client.environment}")
print(f"API URL: {client.api_url}")

Endpoint Overrides (Per Service)

By default, all SDK APIs use one base URL (client.api_url). You can override this globally or per service when you need mixed routing (for example, local orchestrator + remote patients).

Environment variable overrides:

# Global override for all services
export EARL_API_URL="http://localhost:8006/api/v1"

# Per-service overrides (optional)
export EARL_CASES_API_URL="https://earl-api.thelumos.xyz/api/v1"
export EARL_DIMENSIONS_API_URL="http://localhost:8006/api/v1"
export EARL_PATIENTS_API_URL="https://earl-api.thelumos.xyz/api/v1"
export EARL_PIPELINES_API_URL="http://localhost:8006/api/v1"
export EARL_SIMULATIONS_API_URL="http://localhost:8006/api/v1"

# Optional Auth0 endpoint overrides
export EARL_AUTH0_DOMAIN="dev-f4675lf8h3k0i3me.us.auth0.com"
export EARL_AUTH0_AUDIENCE="https://earl-api.thelumos.xyz"  # or .onlyevals.com / .ai per env

Constructor overrides (take precedence over environment variables):

client = EarlClient(
    client_id="...",
    client_secret="...",
    environment="local",
    service_api_urls={
        "patients": "https://earl-api.thelumos.xyz/api/v1",
        "simulations": "http://localhost:8006/api/v1",
    },
)
print(client.service_api_urls)

Doctor API Configuration

Using EARL's Internal Doctor (Default)

If you don't specify a doctor_config, EARL uses its built-in AI doctor:

pipeline = client.pipelines.create(
    name="internal-doctor-test",
    dimension_ids=["factuality", "empathy"],
    patient_ids=patient_ids,
    # No doctor_config = uses internal doctor
)

Using Your External Doctor API

Test your own doctor API:

from earl_sdk import DoctorApiConfig

# Create external doctor config
doctor_config = DoctorApiConfig.external(
    api_url="https://your-doctor.com/chat",
    api_key="your-secret-key",
)

pipeline = client.pipelines.create(
    name="my-doctor-test",
    dimension_ids=["factuality", "empathy", "safety"],
    patient_ids=patient_ids,
    doctor_config=doctor_config,
)

Validate Your Doctor API First

Before creating a pipeline, you can validate your doctor API is reachable:

try:
    result = client.pipelines.validate_doctor_api(
        api_url="https://your-doctor.com/chat",
        api_key="your-key",
    )
    print(f"✓ {result['message']}")
except ValidationError as e:
    print(f"✗ {e}")

🔐 Client-Driven Mode (VPN/Firewall Safe)

If your doctor API is behind a VPN, firewall, or otherwise unreachable from the cloud, use client-driven mode. In this mode, YOUR code acts as the middleware - you pull patient messages and push doctor responses.

from earl_sdk import EarlClient, DoctorApiConfig

client = EarlClient(
    client_id="your-client-id",
    client_secret="your-secret",
    environment="test",
)

# Step 1: Create a CLIENT-DRIVEN pipeline
pipeline = client.pipelines.create(
    name="vpn-doctor-eval",
    dimension_ids=["factuality", "empathy", "safety"],
    patient_ids=patient_ids,
    doctor_config=DoctorApiConfig.client_driven(),  # <-- Key difference!
    conversation_initiator="doctor",  # or "patient"
)

# Step 2: Start simulation
simulation = client.simulations.create(
    pipeline_name=pipeline.name,
    num_episodes=3,
)

# Step 3: YOUR CODE orchestrates the conversation
import time

max_turns = 6
poll_interval = 5.0

while True:
    sim = client.simulations.get(simulation.id)
    if sim.status.value in ["completed", "failed"]:
        print(f"Simulation {sim.status.value}!")
        break
    
    episodes = client.simulations.get_episodes(simulation.id)
    
    for ep in episodes:
        if ep["status"] != "awaiting_doctor":
            continue
        
        # Fetch full episode to get dialogue history
        full_ep = client.simulations.get_episode(simulation.id, ep["episode_id"])
        dialogue = full_ep.get("dialogue_history", [])
        
        # Get patient's message (if any)
        if dialogue and dialogue[-1]["role"] == "patient":
            patient_msg = dialogue[-1]["content"]
            print(f"Patient: {patient_msg[:80]}...")
        
        # Call YOUR doctor API (behind VPN, on localhost, etc.)
        doctor_response = call_your_doctor_api(dialogue)  # Your implementation
        
        # Submit doctor's response back to Earl
        updated_ep = client.simulations.submit_response(
            simulation.id,
            ep["episode_id"],
            doctor_response,
        )
        print(f"Doctor: {doctor_response[:80]}...")
    
    time.sleep(poll_interval)

# Step 4: Get complete report
report = client.simulations.get_report(simulation.id)
print(f"Score: {report['summary']['average_score']:.2f}/4")

Key Points:

• Use DoctorApiConfig.client_driven() - Earl won't call any doctor API
• Client-driven is only for external doctors (behind VPN/firewall). You cannot use it with internal doctor.
• Poll episodes with get_episodes() to see status
• Fetch individual episodes with get_episode() to get full dialogue_history
• Submit responses with submit_response()
• Episode status will be awaiting_doctor when it's your turn

Doctor API Contract

Your doctor API must accept POST requests with this format:

{
  "messages": [
    {"role": "user", "content": "Patient message..."},
    {"role": "assistant", "content": "Previous doctor response..."}
  ],
  "patient_context": {"patient_id": "..."}
}

And return (any of these formats):

{
  "response": "Doctor's response text..."
}

Or OpenAI-compatible format:

{
  "choices": [
    {"message": {"content": "Doctor's response text..."}}
  ]
}

Authentication: Earl sends credentials in BOTH headers for compatibility:

• X-API-Key: your-key
• Authorization: Bearer your-key

Your API can check whichever header you prefer.

Conversation Flow Configuration

You can configure who initiates the conversation and how long it lasts:

Patient-Initiated (Default)

The patient sends the first message describing their symptoms. This is the typical telemedicine flow:

pipeline = client.pipelines.create(
    name="telemedicine-eval",
    dimension_ids=["factuality", "empathy"],
    patient_ids=patient_ids,
    conversation_initiator="patient",  # Default
)
# Patient: "I've been having headaches for a week..."
# Doctor: "I'm sorry to hear that. Can you describe the pain?"

Doctor-Initiated

The doctor sends the first message (greeting/opening). Useful for proactive care or follow-up scenarios:

pipeline = client.pipelines.create(
    name="proactive-care-eval",
    dimension_ids=["empathy", "thoroughness"],
    patient_ids=patient_ids,
    conversation_initiator="doctor",
)
# Doctor: "Hello, I'm Dr. Smith. What brings you in today?"
# Patient: "I've been feeling dizzy lately..."

Maximum Conversation Turns

Control how long conversations can last with max_turns (1-50, default 10):

# Short conversations (quick evaluations)
pipeline = client.pipelines.create(
    name="quick-eval",
    dimension_ids=["factuality"],
    patient_ids=patient_ids,
    max_turns=5,  # End after 5 turns
)

# Longer, more thorough conversations
pipeline = client.pipelines.create(
    name="detailed-eval",
    dimension_ids=["thoroughness", "factuality", "empathy"],
    patient_ids=patient_ids,
    max_turns=30,  # Allow up to 30 turns
)

The patient will naturally indicate they need to leave as the conversation approaches the turn limit.

Parameter	Range	Default	Description
max_turns	1-50	10	Maximum conversation turns before ending

Check Pipeline's Conversation Settings

pipeline = client.pipelines.get("my-pipeline")
print(f"Initiator: {pipeline.conversation_initiator}")  # "patient" or "doctor"
print(f"Max turns: {pipeline.conversation.max_turns}")  # 1-50

Working with Simulations

Start a Simulation

simulation = client.simulations.create(
    pipeline_name="my-pipeline",
    num_episodes=5,      # Number of patient conversations
    parallel_count=2,    # Parallel episodes (1-10)
)

print(f"Simulation ID: {simulation.id}")
print(f"Status: {simulation.status}")

Track Progress

# Get current status
sim = client.simulations.get(simulation_id)
print(f"Progress: {sim.completed_episodes}/{sim.total_episodes}")
print(f"Status: {sim.status}")

# Wait with progress callback
def on_progress(sim):
    print(f"  {sim.completed_episodes}/{sim.total_episodes} completed")

completed = client.simulations.wait_for_completion(
    simulation_id,
    poll_interval=5.0,   # Check every 5 seconds
    timeout=600.0,       # 10 minute timeout
    on_progress=on_progress,
)

Get Episode Details

# Get all episodes (summary view - no dialogue_history for efficiency)
episodes = client.simulations.get_episodes(simulation_id)
for ep in episodes:
    print(f"Episode {ep['episode_number']}: {ep['status']}")
    if ep['status'] == 'completed':
        print(f"  Score: {ep['total_score']:.2f}/4")

# Get single episode with FULL dialogue history
# Use this for client-driven mode to see conversation state
episode = client.simulations.get_episode(simulation_id, episode_id)
print(f"Status: {episode['status']}")  # e.g., "awaiting_doctor"

for turn in episode.get('dialogue_history', []):
    print(f"  {turn['role']}: {turn['content'][:50]}...")

Note: The list endpoint (get_episodes) returns a summary without dialogue_history for performance. To get the full dialogue, fetch individual episodes with get_episode().

Get Report

Get a complete report with all episode data, dialogue history, and judge feedback in one call:

report = client.simulations.get_report(simulation_id)

# Summary statistics
summary = report['summary']
print(f"Completed: {summary['completed']}/{summary['total_episodes']}")
print(f"Average Score: {summary['average_score']:.2f}/4")

# Per-dimension breakdown
print("\nDimension Scores:")
for dim_id, stats in report.get('dimension_scores', {}).items():
    print(f"  {dim_id}: avg={stats['average']:.2f}, min={stats['min']}, max={stats['max']}")

# All episodes with full details
print("\nEpisodes:")
for ep in report['episodes']:
    print(f"\n  Episode {ep['episode_number']}: {ep['patient_name']}")
    print(f"    Score: {ep['total_score']}")
    print(f"    Dialogue ({ep['dialogue_turns']} turns):")
    for turn in ep.get('dialogue_history', [])[:3]:  # First 3 turns
        role = turn['role'].upper()
        content = turn['content'][:60] + "..." if len(turn['content']) > 60 else turn['content']
        print(f"      {role}: {content}")

Rate Limits

API calls are rate-limited per organization. You can check your current limits programmatically:

# Get your organization's rate limits
limits = client.rate_limits.get()

print(f"Organization: {limits['organization_id']}")
print(f"Per minute: {limits['limits']['per_minute']}")
print(f"Per hour: {limits['limits']['per_hour']}")
print(f"Per day: {limits['limits']['per_day']}")

# Category-specific limits
print("\nEffective limits by category:")
for category, limit in limits['effective_limits'].items():
    print(f"  {category}: {limit}/min")

# Quick check for a specific category
sim_limit = client.rate_limits.get_effective_limit("simulations")
print(f"\nSimulations limit: {sim_limit}/min")

Rate Limit Headers

Every API response includes rate limit headers:

Header	Description
X-RateLimit-Limit	Maximum requests allowed in current window
X-RateLimit-Remaining	Requests remaining in current window
X-RateLimit-Reset	Unix timestamp when the window resets

When you exceed the limit, you'll receive HTTP 429 Too Many Requests.

Error Handling

from earl_sdk import EarlClient
from earl_sdk.exceptions import (
    EarlError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
    SimulationError,
)

try:
    report = client.simulations.get_report("invalid-id")
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
except AuthorizationError as e:
    print(f"Access denied: {e.message}")
except NotFoundError as e:
    print(f"Not found: {e}")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except SimulationError as e:
    print(f"Simulation {e.simulation_id} failed: {e.message}")
except ServerError as e:
    print(f"Server error: {e.message}")
except EarlError as e:
    print(f"Unexpected error: {e}")

API Reference

EarlClient

Main entry point for the SDK.

client = EarlClient(
    client_id="...",
    client_secret="...",
    organization="org_xxx",
    environment="test",  # or "prod"
)

# Properties
client.environment   # Current environment
client.api_url       # API URL
client.organization  # Organization ID

# Test connection
client.test_connection()  # Returns True or raises

DimensionsAPI

# List all dimensions
dimensions = client.dimensions.list(include_custom=True)

# Get a specific dimension
dimension = client.dimensions.get("factuality")

# Create custom dimension
dim = client.dimensions.create(
    name="Medical Accuracy",
    description="How accurate is the medical information",
    category="quality",
    weight=1.0,
)

PatientsAPI

# List patients
patients = client.patients.list(
    difficulty="medium",  # easy, medium, hard
    limit=100,
    offset=0,
)

# Get a specific patient
patient = client.patients.get("patient_id")

PipelinesAPI

# List pipelines (summary view)
pipelines = client.pipelines.list(active_only=True)

# Get pipeline with FULL details
pipeline = client.pipelines.get("pipeline_name")
print(f"Doctor: {pipeline.doctor_api.type}")  # 'internal' or 'external'
print(f"Patients: {pipeline.patient_ids}")
print(f"Dimensions: {pipeline.dimension_ids}")
print(f"Initiator: {pipeline.conversation.initiator}")  # 'patient' or 'doctor'

# Create pipeline
pipeline = client.pipelines.create(
    name="my-pipeline",
    dimension_ids=["factuality", "empathy"],
    patient_ids=["patient1", "patient2"],
    doctor_config=DoctorApiConfig.external(...),
    description="My evaluation pipeline",
    validate_doctor=True,  # Validate API before creating
)

# Validate external doctor API
result = client.pipelines.validate_doctor_api(
    api_url="https://...",
    api_key="...",
)

# Update pipeline
client.pipelines.update(
    "pipeline_name",
    description="Updated description",
)

# Delete pipeline
client.pipelines.delete("pipeline_name")

SimulationsAPI

# List simulations
simulations = client.simulations.list(
    pipeline_id="my-pipeline",
    status=SimulationStatus.COMPLETED,
    limit=50,
)

# Get simulation
sim = client.simulations.get("simulation_id")

# Create simulation
sim = client.simulations.create(
    pipeline_name="my-pipeline",
    num_episodes=5,
    parallel_count=2,
)

# Wait for completion
completed = client.simulations.wait_for_completion(
    simulation_id,
    poll_interval=5.0,
    timeout=600.0,
    on_progress=lambda s: print(f"{s.progress:.0%}"),
)

# Get episodes
episodes = client.simulations.get_episodes(
    simulation_id,
    include_dialogue=True,
)

# Get single episode
episode = client.simulations.get_episode(simulation_id, episode_id)

# Get complete report with all details
report = client.simulations.get_report(simulation_id)

# Cancel simulation
client.simulations.cancel(simulation_id)

# Client-driven mode: submit doctor response
updated_episode = client.simulations.submit_response(
    simulation_id,
    episode_id,
    message="Doctor's response text...",
)

RateLimitsAPI

# Get all rate limit info
limits = client.rate_limits.get()
# Returns: {
#   "organization_id": "org_xxx",
#   "limits": {"per_minute": 60, "per_hour": 1000, "per_day": 10000},
#   "category_limits": {"evaluations": 10, "pipelines": 60, ...},
#   "effective_limits": {"evaluations": 10, "pipelines": 60, ...},
#   "headers_info": {...},
# }

# Get effective limit for a specific category
limit = client.rate_limits.get_effective_limit("simulations")  # Returns int

Models

Simulation

sim.id                  # Simulation ID
sim.pipeline_name        # Pipeline name
sim.status              # SimulationStatus enum
sim.total_episodes      # Total episodes
sim.completed_episodes  # Completed episodes
sim.progress            # Progress ratio (0.0-1.0)
sim.error_message       # Error message if failed
sim.summary             # Summary dict if completed

SimulationStatus

from earl_sdk import SimulationStatus

SimulationStatus.PENDING
SimulationStatus.RUNNING
SimulationStatus.COMPLETED
SimulationStatus.FAILED
SimulationStatus.CANCELLED

Understanding Simulation & Episode Status

When running simulations (especially in client-driven mode), use these statuses to track progress.

Simulation Statuses

Status	Description	What to do
running	Simulation is active, episodes are being processed	Keep polling/orchestrating
completed	All episodes finished successfully	Fetch report via get_report()
failed	Simulation failed (critical error)	Check error field
stopped	Simulation was cancelled	N/A

Episode Statuses

Status	Description	Client-Driven Action
pending	Episode created, waiting to start	Wait for orchestrator to initialize
awaiting_doctor	Waiting for YOUR response	Fetch dialogue, call your doctor, submit response
conversation	Dialogue ongoing (internal/external modes)	N/A (orchestrator handles)
judging	Conversation ended, judge evaluating	Wait for completion
completed	✅ Done! Scores available	Read total_score, judge_scores
failed	❌ Error occurred	Check error field for details

Client-Driven Status Flow

┌─────────────────────────────────────────────────────────────────────────┐
│                        CLIENT-DRIVEN WORKFLOW                            │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  1. SIMULATION STARTS                                                   │
│     └─> Simulation status: "running"                                    │
│         └─> Episodes created with status: "pending"                     │
│                                                                         │
│  2. EPISODES INITIALIZE                                                 │
│     └─> Episode status: "awaiting_doctor"                               │
│         └─> If patient initiates: dialogue_history has patient message  │
│         └─> If doctor initiates: dialogue_history is empty              │
│                                                                         │
│  3. YOUR CODE ORCHESTRATES (repeat until done)                          │
│     ┌──────────────────────────────────────────────────────────────┐   │
│     │ a) Poll: get_episode() to see dialogue_history               │   │
│     │ b) Call YOUR doctor API with the conversation               │   │
│     │ c) Submit: submit_response() with doctor's reply            │   │
│     │ d) Orchestrator calls patient, updates dialogue             │   │
│     │ e) Status returns to "awaiting_doctor" for next turn        │   │
│     └──────────────────────────────────────────────────────────────┘   │
│                                                                         │
│  4. CONVERSATION ENDS (doctor says goodbye or max turns)               │
│     └─> Episode status: "judging"                                       │
│                                                                         │
│  5. JUDGE EVALUATES                                                     │
│     └─> Episode status: "completed" (or "failed" if error)             │
│         └─> total_score, judge_scores, judge_feedback available        │
│                                                                         │
│  6. ALL EPISODES DONE                                                   │
│     └─> Simulation status: "completed"                                  │
│         └─> summary.average_score available                             │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Checking Status in Code

# Poll simulation status
sim = client.simulations.get(simulation_id)
print(f"Simulation: {sim.status.value}")  # "running", "completed", etc.

# Get episode list (for IDs and basic status)
episodes = client.simulations.get_episodes(simulation_id)

for ep in episodes:
    ep_id = ep["episode_id"]
    status = ep["status"]
    
    if status == "pending":
        print(f"Episode {ep_id}: Initializing...")
        
    elif status == "awaiting_doctor":
        # Fetch full episode to get dialogue
        full_ep = client.simulations.get_episode(simulation_id, ep_id)
        dialogue = full_ep["dialogue_history"]
        
        if dialogue:
            last_msg = dialogue[-1]
            print(f"Episode {ep_id}: Patient said: {last_msg['content'][:50]}...")
        else:
            print(f"Episode {ep_id}: Doctor should initiate conversation")
        
        # YOUR CODE: Call doctor API, then submit response
        doctor_reply = call_your_doctor(dialogue)
        client.simulations.submit_response(simulation_id, ep_id, doctor_reply)
        
    elif status == "judging":
        print(f"Episode {ep_id}: Being evaluated by judge...")
        
    elif status == "completed":
        print(f"Episode {ep_id}: Score = {ep.get('total_score', 'N/A')}")
        
    elif status == "failed":
        print(f"Episode {ep_id}: FAILED - {ep.get('error', 'Unknown error')}")

Determining When Everything is Done

import time

while True:
    sim = client.simulations.get(simulation_id)
    
    # Check simulation-level status
    if sim.status.value == "completed":
        print("✓ All episodes completed and judged!")
        break
    elif sim.status.value == "failed":
        print(f"✗ Simulation failed: {sim.error_message}")
        break
    
    # Or check episode-level
    episodes = client.simulations.get_episodes(simulation_id)
    all_done = all(ep["status"] in ["completed", "failed"] for ep in episodes)
    
    if all_done:
        print("✓ All episodes finished!")
        break
    
    time.sleep(10)  # Poll every 10 seconds

DoctorApiConfig

from earl_sdk import DoctorApiConfig

# Internal doctor (EARL's built-in AI)
config = DoctorApiConfig.internal()
config = DoctorApiConfig.internal(prompt="Custom system prompt")

# External doctor (your API - Earl calls it directly)
config = DoctorApiConfig.external(
    api_url="https://your-api.com/chat",
    api_key="your-key",
    auth_type="bearer",  # "bearer" (default) or "api_key"
)

# External doctor with X-API-Key header (custom APIs)
config = DoctorApiConfig.external(
    api_url="https://custom-api.com/generate",
    api_key="your-key",
    auth_type="api_key",  # Uses X-API-Key header instead of Authorization: Bearer
)

# Client-driven (YOU control the conversation loop)
# Use when your EXTERNAL doctor API is behind VPN/firewall
# NOTE: client_driven is NOT available with internal doctor
config = DoctorApiConfig.client_driven()

# Check the mode
print(config.type)           # "internal", "external", or "client_driven"
print(config.is_client_driven)  # True/False

Score Scale

Evaluation scores are on a 1-4 scale:

Score	Meaning
1	Poor
2	Fair
3	Good
4	Excellent

Testing

Interactive UI

The quickest way to explore and test is through the interactive UI:

# Install with UI dependencies
pip install -e ".[ui]"

# Launch
earl-ui

# Or via make (from project root)
make sdk-ui

SDK Integration Tests

SDK integration tests are in the tests/ directory. Credentials can be passed via CLI or environment variables.

Test Internal Doctor (Earl's Built-in)

# Test with 2 patients
python3 tests/test_doctors.py --env test --doctor internal --patients 2 --wait \
    --client-id "your-client-id" \
    --client-secret "your-client-secret"

Test External Doctor (Your API)

python3 tests/test_doctors.py --env test --doctor external --patients 3 --wait \
    --client-id "your-client-id" \
    --client-secret "your-client-secret" \
    --doctor-url "https://your-api.com/v1/chat/completions" \
    --doctor-key "your-api-key"

Test Client-Driven Mode (VPN/Firewall)

# With mock doctor (for testing the workflow)
python3 tests/test_client_driven.py --env test \
    --client-id "your-client-id" \
    --client-secret "your-client-secret"

# With your local doctor API
python3 tests/test_client_driven.py --env test \
    --client-id "your-client-id" \
    --client-secret "your-client-secret" \
    --local-doctor-url "http://localhost:8080/chat" \
    --local-doctor-key "your-key"

List Patients Only

python3 tests/test_doctors.py --env test --list-only \
    --client-id "your-client-id" \
    --client-secret "your-client-secret"

Using Environment Variables (Alternative)

# Set credentials once
export EARL_CLIENT_ID="your-client-id"
export EARL_CLIENT_SECRET="your-client-secret"

# Then run tests without --client-id/--client-secret
python3 tests/test_doctors.py --env test --doctor internal --patients 2 --wait

Support

• Email: [e-mail]

License

MIT License - see LICENSE file for details.