Dispersl Python SDK

Flexible workflow automation with plug-and-play agents for Python. Behavioral feature parity with @codefundi/dispersl-sdk (TypeScript).

Install

pip install dispersl-sdk

Requirements

• Python >=3.9
• Async runtime (asyncio)

Quick Start

import asyncio

from dispersl import AgenticExecutor, AsyncDisperslClient


async def main() -> None:
    client = AsyncDisperslClient(
        base_url="https://api.dispersl.com/v1",
        api_key="YOUR_API_KEY",
        timeout_s=120.0,
        retry_attempts=3,
    )

    executor = AgenticExecutor(client)
    out = await executor.run_plan_and_agent_loop(
        prompt="Plan and implement a production webhook pipeline",
        agent_choices="auto",  # or ["architect", "security-auditor", "release-manager"]
        execution_sequence="sequential",  # or "parallel" for concurrent agent execution
    )

    print(out["task_id"], len(out["events"]), len(out["tool_results"]))
    await client.aclose()


asyncio.run(main())

SDK Capabilities

• Async HTTP client with bearer auth, retries, timeout support, and structured error mapping.
• Optional codefundi_api_key → x-codefundi-key metering header; session state headers (x-dispersl-state, x-dispersl-turn-seq).
• Full endpoint coverage for agent/completion, agent/plan, and agent lifecycle APIs.
• Incremental NDJSON stream parser (NdjsonParser) with split-buffer handling and parse errors.
• NDJSON chunk normalization (inline tool_calls in content, top-level tool_calls → tools) via StreamContentNormalizer.
• Stream turn parsing (parse_agent_stream): one API stream → N local tool executions → grouped results.
• Stateful grouped tool feedback (build_grouped_tool_feedback_prompt) — no previous-assignment echo.
• Loop detection (soft/hard thresholds, ping-pong, result-aware signatures).
• Handover parser supporting nested and double-serialized tool arguments; control tools not executed locally.
• Sequential and parallel agent execution modes for flexible workflow orchestration.
• Task continuation support via task_id for multi-phase workflows.
• MCP config loading from .dispersl/mcp.json with env interpolation, validation, catalog search/describe meta tools.
• Agentic execution loop with plan-to-agent transitions, tool execution, and end-session detection (max_loops default 25).

Client API Surface

Agent execution endpoints

Agent plan choices

agent_plan accepts:

• "auto" for automatic custom-agent selection
• list[str] for explicit custom agent name_id values

When "auto" is passed, the SDK normalizes the request payload to ["auto"] for API compatibility.

Execution modes

agent_plan and run_plan_and_agent_loop support execution sequence control:

• "sequential" (default): agents execute one after another
• "parallel": multiple agents execute concurrently when handed over from plan

For parallel execution, use parallel_concurrency to limit simultaneous agent runs.

Resource endpoints

Execution Loop Behavior

AgenticExecutor.run_plan_and_agent_loop includes:

• plan-first execution and automatic transition to selected specialist agent
• execution sequence control (execution_sequence: "sequential" or "parallel")
• parallel concurrency limit (parallel_concurrency)
• handover propagation; control tools (end_session, finish_task, handover_task) are not passed to tool_executor
• stateful grouped continuation prompts via tool_feedback (no turn-text / previous-assignment echo)
• soft loop → extra_directive; hard loop → synthetic error chunk + stop
• optional custom tool runner (tool_executor)
• deterministic termination with max_loops (default 25)
• returned payload: task_id, events, tool_results, execution_sequence, workflow_complete

Direct Single-Agent Completion Loop

executor = AgenticExecutor(client)
result = await executor.run_agent_completion_loop(
    name_id="architect",
    prompt="Review this backend design and produce a migration plan",
    max_loops=25,
)

System-one agents (Jev)

Use agent_model for system-one specialists (typesafe/jev-1.13) while model stays the planner langage model. Pass build_system_one_prompt so each handover string becomes {state, questions}. Chunks include model_type and system_one. Sequential hops still follow handover_task; parallel specialists stop on handover_task.

result = await executor.run_plan_and_agent_loop(
    prompt="Plan the vote",
    agent_choices=["market-research-specialist"],
    model="stealth/ox-alpha",
    agent_model="typesafe/jev-1.13",
    build_system_one_prompt=lambda handover_text: {
        "state": {"question": handover_text},
        "questions": {
            "action": {
                "type": "choice",
                "instructions": "Pick an action",
                "criteria": {"BUY_UP": "Buy up", "HOLD": "Hold"},
            }
        },
    },
)

Task Continuation

first_run = await executor.run_plan_and_agent_loop(
    prompt="Design the system architecture",
    agent_choices="auto",
)

result = await executor.run_plan_and_agent_loop(
    prompt="Now implement the core modules",
    agent_choices="auto",
    task_id=first_run["task_id"],
)

Core Models and Helpers

Security / Session State Env Vars

Error Model

Development

python -m pip install -e ".[dev]"
python -m ruff check src tests
python -m ruff format --check src tests
python -m mypy src
python -m pytest -q
python -m build

Example Quickstarts

End-to-end quickstarts live in root examples/py:

• examples/py/plan_handover_loop.py
• examples/py/single_agent_completion.py
• examples/py/task_insight_progress.py
• examples/py/agent_lifecycle_and_stats.py
• examples/py/mcp_custom_agent_flow.py

Release

• Package name: dispersl-sdk
• Version: 0.1.6
• Python release workflow: .github/workflows/release-python.yml
• Trigger: push tag py-v*