Docs

Client SDK Guide

Everything you need to integrate the CROWDio SDK into your own Python code.

Installation

The SDK ships as part of the developer_sdk package in the CROWDio repository. After installing dependencies:

pip install -r requirements.txt

Import from the package:

from developer_sdk import connect, map, run, submit, get, pipeline, disconnect

Or use the client class directly:

from developer_sdk.client import CrowdComputeClient

Public API Reference

Function Description
await connect(host, port=9000) Open WebSocket connection to Foreman
await map(func, iterable, **kwargs) Distribute function over iterable; blocks until all results received
await run(func, *args, **kwargs) Run a single function call on a remote worker
await submit(func, iterable, **kwargs) Non-blocking submit; returns job_id
await get(job_id, timeout=None) Poll/wait for results of a previously submitted job
await pipeline(stages, dependency_map=None, **kwargs) Submit a multi-stage pipeline
await disconnect() Close connection cleanly

Example 1 — Basic map()

The simplest pattern: distribute a function over a list of arguments.

import asyncio
from developer_sdk import connect, map, disconnect


def square(x):
    return x * x


async def main():
    await connect("localhost", 9000)
    results = await map(square, [1, 2, 3, 4, 5])
    print(results)  # [1, 4, 9, 16, 25]
    await disconnect()


if __name__ == "__main__":
    asyncio.run(main())

!!! note “map() is blocking” map() waits until all tasks are complete and returns the ordered results. For fire-and-forget workflows, use submit() + get().

Example 2 — run() (Single Call)

Dispatch a single function call to a remote worker.

import asyncio
from developer_sdk import connect, run, disconnect


def greet(name):
    return f"Hello, {name}!"


async def main():
    await connect("localhost", 9000)
    result = await run(greet, "World")
    print(result)  # Hello, World!
    await disconnect()


if __name__ == "__main__":
    asyncio.run(main())

Example 3 — Async submit() / get()

Submit a job without waiting, then poll for results later. Useful for long-running jobs or batching submissions.

import asyncio
from developer_sdk import connect, submit, get, disconnect


def heavy(x):
    return x ** 3


async def main():
    await connect("localhost", 9000)

    # Fire-and-forget submit
    job_id = await submit(heavy, [1, 2, 3, 4, 5])
    print(f"Submitted job: {job_id}")

    # ... do other work here ...

    # Wait for results (60s timeout)
    results = await get(job_id, timeout=60)
    print(results)  # [1, 8, 27, 64, 125]

    await disconnect()


if __name__ == "__main__":
    asyncio.run(main())

Example 4 — pipeline()

Multi-stage pipelines where the output of one stage feeds the next.

import asyncio
from developer_sdk import connect, pipeline, disconnect


def stage_a(x):
    return x * 2


def stage_b(x):
    return x + 1


async def main():
    await connect("localhost", 9000)

    stages = [
        {"func": stage_a, "args_list": [1, 2, 3]},
        {
            "func": stage_b,
            "args_list": [None, None, None],
            "pass_upstream_results": True,  # receives stage_a outputs
        },
    ]

    results = await pipeline(stages)
    print(results)  # stage_b([2, 4, 6]) → [3, 5, 7]

    await disconnect()


if __name__ == "__main__":
    asyncio.run(main())

Pipeline Stage Keys

Key Required Description
func ✅ Callable to distribute
args_list ✅ One argument per task slot
pass_upstream_results Optional If True, previous stage outputs are injected as args
dependency_map Optional Custom inter-stage dependency definitions

Using the CrowdComputeClient Directly

For advanced use cases requiring multiple concurrent connections:

import asyncio
from developer_sdk.client import CrowdComputeClient


async def main():
    client = CrowdComputeClient()
    await client.connect("localhost", 9000)

    results = await client.map(lambda x: x ** 2, range(10))
    print(results)

    await client.disconnect()


asyncio.run(main())

Error Handling

import asyncio
from developer_sdk import connect, map, disconnect


def might_fail(x):
    if x == 3:
        raise ValueError("bad input")
    return x


async def main():
    await connect("localhost", 9000)
    try:
        results = await map(might_fail, [1, 2, 3, 4, 5])
    except Exception as e:
        print(f"Job failed: {e}")
    finally:
        await disconnect()


asyncio.run(main())

!!! warning “Partial failures” If any task fails and has exhausted retries, the job is marked FAILED and job_error is returned to the client. Configure retries using the @crowdio.task decorator.

Next Steps