Quick reference for example files demonstrating SDK features.

# Install
pip install conductor-python httpx

# Configure
export CONDUCTOR_SERVER_URL="http://localhost:8080/api"

# Run end-to-end example
python examples/workers_e2e.py

Key Concepts:

• def → TaskRunner (ThreadPoolExecutor)
• async def → AsyncTaskRunner (pure async/await, single event loop)
• One process per worker (automatic selection)

from conductor.client.context.task_context import TaskInProgress
from typing import Union

@worker_task(task_definition_name='batch_job')
def process_batch(batch_id: str) -> Union[dict, TaskInProgress]:
    ctx = get_task_context()

    if ctx.get_poll_count() < 5:
        # More work - extend lease
        return TaskInProgress(callback_after_seconds=30)

    return {'status': 'completed'}

See: task_context_example.py, worker_example.py

See agentic_workflows/ for the full set of AI agent examples.

Access metrics: curl http://localhost:8000/metrics

Complete working examples demonstrating 100% API coverage for major SDK features.

Scenario: E-commerce platform with departments, teams, and role-based access control.

Features:

• User, group, and application management
• Custom roles with fine-grained permissions
• Resource access control and audit trails
• Automatic cleanup (use --no-cleanup to keep resources)

python examples/authorization_journey.py

Scenario: E-commerce order processing with scheduled batch workflows.

Features:

• Schedule CRUD operations
• Cron expressions with timezone support
• Pause/resume schedules
• Execution history and monitoring

python examples/schedule_journey.py

Scenario: Online education platform with complex workflow orchestration.

Features:

• Task and workflow definition management
• Version control and tagging
• Rate limiting and monitoring
• Complex workflow patterns (SWITCH, FORK_JOIN, DECISION)

python examples/metadata_journey.py

Scenario: AI-powered customer service with managed prompt templates.

Features:

• Prompt template CRUD operations
• Multi-language support
• Testing with AI models
• Version management and tagging

python examples/prompt_journey.py

Complete RAG (Retrieval Augmented Generation) pipeline example:

# 1. Install dependencies
pip install conductor-python "markitdown[pdf]"

# 2. Configure (requires Orkes Conductor with AI/LLM support)
#    - Vector DB integration named "postgres-prod" (pgvector)
#    - LLM provider named "openai" with a valid API key
export CONDUCTOR_SERVER_URL="http://localhost:7001/api"

# 3. Run RAG workflow
python examples/rag_workflow.py examples/goog-20251231.pdf "What were Google's total revenues?"

Pipeline: convert_to_markdown → LLM_INDEX_TEXT → WAIT → LLM_SEARCH_INDEX → LLM_CHAT_COMPLETE

Features:

• Document conversion (PDF, Word, Excel → Markdown via markitdown)
• Vector database ingestion into pgvector with OpenAI text-embedding-3-small embeddings
• Semantic search with configurable result count
• Context-aware answer generation with gpt-4o-mini

MCP (Model Context Protocol) agent example:

# 1. Install MCP weather server
pip install mcp-weather-server

# 2. Start MCP server
python3 -m mcp_weather_server \
  --mode streamable-http \
  --host localhost \
  --port 3001 \
  --stateless

# 3. Run AI agent
export OPENAI_API_KEY="your-key"
export ANTHROPIC_API_KEY="your-key"
python examples/agentic_workflows/mcp_weather_agent.py "What's the weather in Tokyo?"

# Or simple mode (direct tool call):
python examples/agentic_workflows/mcp_weather_agent.py "Temperature in New York" --simple

Features:

• MCP tool discovery
• LLM-based planning (agent decides which tool to use)
• Tool execution via HTTP/Streamable transport
• Natural language response generation

# 1. Basic workers (5 min)
python examples/workers_e2e.py

# 2. Long-running tasks (5 min)
python examples/task_context_example.py

# 3. Configuration (5 min)
python examples/worker_configuration_example.py

# 4. Workflows (10 min)
python examples/dynamic_workflow.py

# 5. AI/LLM Workflows (15 min)
python examples/agentic_workflows/llm_chat.py
python examples/rag_workflow.py examples/goog-20251231.pdf "What were Google's total revenues?"

# 6. Monitoring (5 min)
python examples/metrics_example.py
curl http://localhost:8000/metrics

examples/
├── Core Workers
│   ├── workers_e2e.py                  # ⭐ Start here
│   ├── worker_example.py               # Comprehensive patterns
│   ├── worker_configuration_example.py # Env var configuration
│   ├── task_context_example.py         # Long-running tasks
│   ├── task_workers.py                 # Dataclass patterns
│   └── pythonic_usage.py               # Pythonic decorators
│
├── Workflows
│   ├── dynamic_workflow.py             # Workflow creation
│   ├── workflow_ops.py                 # Workflow management
│   ├── workflow_status_listner.py      # Workflow events
│   └── test_workflows.py               # Unit tests
│
├── AI/LLM Workflows
│   ├── rag_workflow.py                 # RAG pipeline (markitdown + pgvector)
│   └── agentic_workflows/             # Agentic AI examples
│       ├── llm_chat.py                # Multi-turn LLM chat
│       ├── llm_chat_human_in_loop.py  # Interactive chat with WAIT
│       ├── multiagent_chat.py         # Multi-agent debate
│       ├── function_calling_example.py # LLM function calling
│       └── mcp_weather_agent.py       # MCP tool calling agent
│
├── Monitoring
│   ├── metrics_example.py              # Prometheus metrics
│   ├── event_listener_examples.py      # Custom listeners
│   └── task_listener_example.py        # Task events
│
├── Advanced
│   ├── task_configure.py               # Task definitions
│   ├── kitchensink.py                  # All features
│   ├── shell_worker.py                 # Shell commands
│   └── untrusted_host.py               # SSL handling
│
├── API Journeys
│   ├── authorization_journey.py        # ⭐ All 49 authorization APIs
│   ├── schedule_journey.py             # ⭐ All 15 schedule APIs
│   ├── metadata_journey.py             # ⭐ All 21 metadata APIs
│   └── prompt_journey.py               # ⭐ All 8 prompt APIs
│
├── helloworld/                         # Simple examples
│   ├── greetings_worker.py
│   ├── greetings_workflow.py
│   └── helloworld.py
│
├── user_example/                       # HTTP + dataclass
│   ├── models.py
│   └── user_workers.py
│
├── worker_discovery/                   # Auto-discovery
│   ├── my_workers/
│   └── other_workers/
│
└── orkes/                             # Orkes-specific features
    ├── vector_db_helloworld.py        # Vector DB operations
    ├── agentic_workflow.py            # AI agent (AIOrchestrator)
    ├── http_poll.py
    ├── sync_updates.py
    └── wait_for_webhook.py

Multiprocess - one process per worker with automatic runner selection:

# Sync worker → TaskRunner (ThreadPoolExecutor)
@worker_task(task_definition_name='cpu_task', thread_count=4)
def cpu_task(data: dict):
    return expensive_computation(data)

# Async worker → AsyncTaskRunner (event loop, 67% less memory)
@worker_task(task_definition_name='api_task', thread_count=50)
async def api_task(url: str):
    async with httpx.AsyncClient() as client:
        return await client.get(url)

# Required
export CONDUCTOR_SERVER_URL="http://localhost:8080/api"

# Optional - Orkes Cloud
export CONDUCTOR_AUTH_KEY="your-key"
export CONDUCTOR_AUTH_SECRET="your-secret"

# Optional - Worker config
export conductor.worker.all.domain=production
export conductor.worker.all.poll_interval_millis=250
export conductor.worker.all.thread_count=20

Workers not polling?

• Check task names match between workflow and @worker_task
• Verify CONDUCTOR_SERVER_URL is correct
• Check auth credentials

Async workers using threads?

• Use async def (not def)
• Check logs for "Created AsyncTaskRunner"

High memory?

• Use async def for I/O tasks (lower memory)
• Reduce worker count or thread_count

• Authorization API - Complete RBAC system (49 APIs)
• Metadata API - Task & workflow definitions (21 APIs)
• Prompt API - AI/LLM prompt templates (8 APIs)
• Schedule API - Workflow scheduling (15 APIs)
• Task Management API - Task operations (11 APIs)
• Workflow API - Workflow operations
• Integration API - AI/LLM provider integrations

• Worker Design - Complete architecture guide
• Worker Configuration - Hierarchical config system

• Python SDK README - SDK overview and installation

Repository: https://github.com/conductor-oss/conductor-python License: Apache 2.0