slimfaas-client

Python client to connect Jobs or virtual functions to SlimFaas via WebSocket. Lets any process receive async requests and publish/subscribe events without exposing an HTTP port.

Links:

PyPI package: https://pypi.org/project/slimfaas-client
GitHub repository: https://github.com/SlimPlanet/SlimFaas
SlimFaas website: https://slimfaas.dev/

Requirements

Python ≥ 3.10
UV as package manager

Installation

uv add slimfaas-client
# or from source
uv pip install -e .

Quick start

import asyncio
from slimfaas_client import (
    SlimFaasClient, SlimFaasClientConfig,
    SubscribeEventConfig, FunctionVisibility,
    AsyncRequest, PublishEvent,
)

async def handle_request(req: AsyncRequest) -> int:
    """Called when SlimFaas sends an async-function request.
    Return an HTTP status code (200 = success, 500 = error, 202 = long processing)."""
    print(f"{req.method} {req.path}{req.query}")
    print(f"Body: {req.body}")
    return 200

async def handle_event(evt: PublishEvent) -> None:
    """Called when SlimFaas publishes a publish-event."""
    print(f"Event '{evt.event_name}': {evt.body}")

async def main():
    config = SlimFaasClientConfig(
        function_name="my-job",
        subscribe_events=[
            SubscribeEventConfig(name="order-created"),
            SubscribeEventConfig(name="order-updated"),
        ],
        default_visibility=FunctionVisibility.PUBLIC,
        number_parallel_request=5,
    )

    async with SlimFaasClient("ws://slimfaas:5003/ws", config) as client:
        client.on_async_request(handle_request)
        client.on_publish_event(handle_event)
        await client.run_forever()

asyncio.run(main())

Full configuration

from slimfaas_client import (
    SlimFaasClientConfig, SubscribeEventConfig, PathVisibilityConfig,
    FunctionVisibility, FunctionTrust,
)

config = SlimFaasClientConfig(
    function_name="my-job",

    # SlimFaas/DependsOn
    depends_on=["other-function"],

    # SlimFaas/SubscribeEvents — each entry may override visibility individually
    subscribe_events=[
        SubscribeEventConfig(name="my-event", visibility=FunctionVisibility.PUBLIC),
        SubscribeEventConfig(name="internal-event"),  # inherits default_visibility
    ],

    # SlimFaas/DefaultVisibility
    default_visibility=FunctionVisibility.PUBLIC,  # or PRIVATE

    # SlimFaas/PathsStartWithVisibility
    paths_start_with_visibility=[
        PathVisibilityConfig(path="/admin", visibility=FunctionVisibility.PRIVATE),
    ],

    # SlimFaas/Configuration
    configuration='{"key": "value"}',

    # SlimFaas/ReplicasStartAsSoonAsOneFunctionRetrieveARequest
    replicas_start_as_soon_as_one_function_retrieve_a_request=True,

    # SlimFaas/NumberParallelRequest
    number_parallel_request=10,

    # SlimFaas/NumberParallelRequestPerPod
    number_parallel_request_per_pod=5,

    # SlimFaas/DefaultTrust
    default_trust=FunctionTrust.TRUSTED,  # or UNTRUSTED
)

Sync streaming (HTTP-over-WebSocket)

from slimfaas_client import SyncRequest

async def handle_sync(req: SyncRequest) -> None:
    body = b'{"status": "ok"}'
    await req.response.start(200, {"Content-Type": ["application/json"]})
    await req.response.write(body)
    await req.response.complete()

client.on_sync_request(handle_sync)

Long-running requests (status 202)

Return 202 to acknowledge the request without completing it yet, then call send_callback when done:

async def handle_long(req: AsyncRequest) -> int:
    asyncio.create_task(process_in_background(req))
    return 202  # "I'll handle it — will call back"

async def process_in_background(req: AsyncRequest) -> None:
    await asyncio.sleep(10)
    await client.send_callback(req.element_id, 200)

Dependency injection

The handlers are plain async functions, so you can close over any dependency you resolved from your DI framework:

# Example with a database session from SQLAlchemy
from sqlalchemy.ext.asyncio import AsyncSession

async def make_handler(session: AsyncSession):
    async def handle_request(req: AsyncRequest) -> int:
        await session.execute(...)  # use the injected session
        return 200
    return handle_request

client.on_async_request(await make_handler(db_session))

Automatic reconnection

The client reconnects automatically after a disconnection. Configure the delay between attempts and the keepalive ping interval:

client = SlimFaasClient(
    "ws://...",
    config,
    reconnect_delay=10.0,
    ping_interval=30.0,  # use 0 to disable keepalive pings
)

Important rules

function_name must not match an existing Kubernetes Deployment name. SlimFaas will reject the registration with a SlimFaasRegistrationError.
All clients sharing the same function_name must have the exact same configuration. Mismatches are rejected on connection.

Development

uv sync --extra dev
uv run pytest

Release files for slimfaas-client 0.84.3

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for slimfaas-client 0.84.3
File	Size	Uploaded
slimfaas_client-0.84.3.tar.gz	79.7 kB	Sep 10, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for slimfaas-client 0.84.3
File	Interpreter	ABI	Platform	Reset
slimfaas_client-0.84.3-py3-none-any.whl 13.2 kB Sep 10, 2026	Python 3	none	any	Details

Total release size: 92.9 kB

Release files / slimfaas_client-0.84.3.tar.gz

Download URL	slimfaas_client-0.84.3.tar.gz
Size	79.7 kB
Tags	Source
SHA-256 checksum How to use checksums	`c865fbe488f0624d43170be312fb1f6b7c43e7a5d885d850dabd402cf1bfdd70`
BLAKE2b-256 checksum How to use checksums	`be8737dcd8a3f23579b96581782d47cba691adc1b9ded69e14dd6c4d390dd705`
Upload date	Sep 10, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.13.14`

Release files / slimfaas_client-0.84.3-py3-none-any.whl

Download URL	slimfaas_client-0.84.3-py3-none-any.whl
Size	13.2 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`1ef0cc0096d50e34a0a4576439376fb61d37040d7c62baac9f0ed962df992dec`
BLAKE2b-256 checksum How to use checksums	`26d0efbbd9740e24b18fb09cea099f6fd8a5e4a8da177dc7f4e532f182e88088`
Upload date	Sep 10, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.13.14`