ModuleX Python SDK

The official Python SDK for the ModuleX AI workflow orchestration platform.

Installation

pip install modulex-python

Quick Start

import asyncio
from modulex import Modulex

async def main():
    async with Modulex(
        api_key="mx_live_...",
        organization_id="your-org-id",
    ) as client:
        # Get current user
        me = await client.auth.me()
        print(f"Hello, {me['username']}!")

        # List workflows
        workflows = await client.workflows.list(status="active")
        for wf in workflows["workflows"]:
            print(f"  {wf['name']}")

asyncio.run(main())

Authentication

Get your API key from the ModuleX Dashboard. Keys use the mx_live_ prefix.

from modulex import Modulex

# Pass API key directly
client = Modulex(api_key="mx_live_...")

# Or use environment variable
import os
client = Modulex(api_key=os.environ["MODULEX_API_KEY"])

Organization Context

Most endpoints require an organization context. Set it at the client level or override per-request:

# Set default org for all requests
client = Modulex(api_key="mx_live_...", organization_id="org-uuid")

# Override for a specific request
workflows = await client.workflows.list(organization_id="other-org-uuid")

Configuration

client = Modulex(
    api_key="mx_live_...",
    organization_id="org-uuid",           # Default organization
    base_url="https://api.modulex.dev",   # API base URL
    timeout=30.0,                         # Request timeout (seconds)
    max_retries=3,                        # Retry count for transient errors
)

Resources

Workflows

# List workflows
workflows = await client.workflows.list(status="active", search="email")

# Auto-paginate all workflows
async for wf in client.workflows.list_all(status="active"):
    print(wf["name"])

# Create workflow
workflow = await client.workflows.create(
    workflow_schema={
        "metadata": {"name": "My Workflow", "version": "1.0"},
        "config": {},
        "state_schema": {"fields": {}},
        "nodes": [],
        "edges": [],
        "entry_point": "start",
    },
    name="My Workflow",
)

# Update & delete
await client.workflows.update("workflow-id", name="New Name", status="active")
await client.workflows.delete("workflow-id")

Executions

# Run a workflow
result = await client.executions.run(
    workflow_id="workflow-uuid",
    input={"messages": [{"role": "user", "content": "Hello!"}]},
)
print(result.run_id)  # typed attribute access (responses are Pydantic models)

# Safely retry a run without double-execution
result = await client.executions.run(
    workflow_id="workflow-uuid",
    input={"messages": [...]},
    idempotency_key="order-4823",  # stable key across retries
)

# Get execution state / resume after interrupt / cancel
state = await client.executions.get_state(thread_id="thread-uuid")
await client.executions.resume(thread_id="thread-uuid", run_id="run-uuid", resume_value="user input")
await client.executions.cancel(run_id="run-uuid", reason="No longer needed")

# Run history (workflow-runs)
runs = await client.executions.list_runs(workflow_id="workflow-uuid", limit=50)
async for run in client.executions.iter_runs(status="succeeded"):  # auto-paginates
    print(run.run_id, run.status)
detail = await client.executions.get_run(run_pk="run-row-id")

Agentic ("direct LLM") chat moved off /workflows/run — use client.assistant.chat(...) instead.

SSE Streaming

The backend carries the event type in the SSE event: field for /chats/stream, and in the JSON data["type"] for workflow/composer/assistant streams. The SDK normalizes both, so event.event always holds the logical type. Streams stop after a terminal event (done/error/cancelled/ interrupted) and heartbeats are filtered by default.

# Listen to workflow execution events
async for event in client.executions.listen(run_id="run-uuid"):
    if event.event == "node_update":
        print(f"Node {event.data['node']}: {event.data.get('output')}")
    elif event.event == "interrupt":
        payload = event.data["data"]  # InterruptEventData is nested under data["data"]
        print(f"Needs input: {payload.get('message')}")
    elif event.is_terminal:
        print(f"Stream ended: {event.event}")

# Listen to chat list updates
async for event in client.chats.stream():
    if event.event == "chat_list_updated":
        print(f"Chat list changed: {event.data}")

# Listen to composer events
async for event in client.composer.listen("chat-id", "run-id"):
    print(f"{event.event}: {event.data}")

Credentials

# Add an API key credential
cred = await client.credentials.create(
    integration_name="openai",
    auth_data={"api_key": "sk-..."},
    display_name="Production OpenAI",
)

# Test a credential
result = await client.credentials.test(cred["credential_id"])
print(f"Valid: {result['is_valid']}")

# List credentials
creds = await client.credentials.list(integration_name="openai")

# Add MCP server
mcp = await client.credentials.create_mcp_server(
    server_url="https://mcp-server.example.com",
    headers={"Authorization": "Bearer ..."},
)

Knowledge Bases

# Create a knowledge base
kb = await client.knowledge.create(
    name="Docs",
    embedding_config={"provider": "openai", "model": "text-embedding-3-small"},
)

# Upload a document
doc = await client.knowledge.upload_document(
    knowledge_base_id=kb["id"],
    file_path="/path/to/doc.pdf",
    metadata={"department": "engineering"},
)

# Search
results = await client.knowledge.search(
    knowledge_base_id=kb["id"],
    query="How does deployment work?",
    top_k=5,
)

# Retrieve RAG context
context = await client.knowledge.retrieve_context(
    knowledge_base_id=kb["id"],
    query="deployment steps",
    max_tokens=2000,
)

Schedules

# Create a cron schedule
schedule = await client.schedules.create(
    workflow_id="workflow-uuid",
    name="Daily Report",
    schedule_type="cron",
    cron_expression="0 9 * * 1-5",
    timezone="America/New_York",
)

# Pause/resume
await client.schedules.pause(schedule["id"])
await client.schedules.resume(schedule["id"])

# View run history
runs = await client.schedules.list_runs(schedule["id"])
stats = await client.schedules.run_stats(schedule["id"], days=30)

Deployments

# Deploy a workflow
deployment = await client.deployments.create(
    workflow_id="workflow-uuid",
    deployment_note="v1.0 release",
)

# Activate a deployment
await client.deployments.activate("workflow-uuid", deployment["id"])

# Deactivate live deployment
await client.deployments.deactivate("workflow-uuid")

Composer

llm is a provider config dict ({integration_name, provider_id, model_id, credential_id?}) — pass a ComposerLLMConfig or an equivalent dict.

from modulex.types import ComposerLLMConfig, YesNoResponse, user_input_request_from_event

result = await client.composer.chat(
    message="Add an LLM node that summarizes the input",
    workflow_id="workflow-uuid",
    llm=ComposerLLMConfig(integration_name="anthropic", provider_id="anthropic", model_id="claude-sonnet-4-20250514"),
)

# Listen, and answer a human-in-the-loop question (HITL) when the run pauses
async for event in client.composer.listen(result.composer_chat_id, result.run_id):
    if event.event == "user_input_request":
        question = user_input_request_from_event(event.data)  # typed UserInputRequest
        await client.composer.resume(
            result.composer_chat_id,
            request_id=question.request_id,
            response=YesNoResponse(answer=True),
            llm={"integration_name": "anthropic", "provider_id": "anthropic", "model_id": "claude-sonnet-4-20250514"},
        )  # returns a NEW run_id — re-subscribe with listen() on it
    elif event.is_terminal:
        break

chats = await client.composer.list(limit=20)        # cursor-paginated
await client.composer.save(result.composer_chat_id)  # or .revert(...)

Assistant (agentic chat)

Shares the HITL contract with the composer. All endpoints are available to any org member.

result = await client.assistant.chat("Summarize my latest runs", llm=ComposerLLMConfig(
    integration_name="openai", provider_id="openai", model_id="gpt-4o-mini",
))
async for event in client.assistant.listen(result.chat_id, result.run_id):
    if event.event == "response_chunk":
        print(event.data.get("data", {}).get("text", ""), end="")
    elif event.is_terminal:
        break

Other Resources

# Organizations
await client.organizations.create(name="My Org")
await client.organizations.invite("user@example.com", role="member")
llms = await client.organizations.llms()

# Dashboard
logs = await client.dashboard.logs(category="CREDENTIALS")
overview = await client.dashboard.analytics_overview()
users = await client.dashboard.users(search="john")

# Subscriptions
plans = await client.subscriptions.organization_plans()
billing = await client.subscriptions.organization_billing()

# Notifications
notifications = await client.notifications.list()

# Integrations
integrations = await client.integrations.browse(type="tool")
providers = await client.integrations.llm_providers()

# System
health = await client.system.health()
timezones = await client.system.timezones()

# API Keys
key = await client.api_keys.create(name="CI/CD Key")
await client.api_keys.revoke(key["id"])

Error Handling

from modulex import (
    Modulex,
    ModulexError,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

try:
    workflow = await client.workflows.get("invalid-id")
except NotFoundError:
    print("Workflow not found")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s (limit={e.limit}, remaining={e.remaining})")
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except ModulexError as e:
    print(f"API error ({e.status_code}): {e.message}")

Usage/billing denials (quota, credit, wallet) are surfaced structurally via BillingError and its subclasses, which expose code, layer, key, current, limit, and reason:

from modulex import BillingError, CreditExhaustedError

try:
    await client.executions.run(workflow_id="wf")
except CreditExhaustedError as e:           # 402, layer="credit"
    print(f"Out of credits: {e.current}/{e.limit}")
except BillingError as e:                   # any quota/credit/wallet denial
    print(f"Denied ({e.layer}/{e.code}): {e.reason}")

Exception Hierarchy

Type Hints

Responses are Pydantic v2 models — use typed attribute access (result.id) or, for compatibility, dict-style access (result["id"]). Unknown fields the backend may add are preserved. All models are importable:

from modulex import SSEEvent
from modulex.types import (
    WorkflowDefinition,
    NodeDefinition,
    EdgeDefinition,
    LLMConfig,
    RunResponse,
    AsyncPage,        # typed auto-pagination (e.g. executions.iter_runs)
    ModulexModel,     # base class for all response models
)

Documentation

For full API documentation, visit docs.modulex.dev.

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines on:

• Setting up the development environment
• Running tests (unit and integration)
• Code style and commit conventions
• Pull request process

License

This project is licensed under the MIT License — see the LICENSE file for details.