sofias-sdk-lite

Build multi-node agents on a declarative graph and run them against RabbitMQ.

sofias-sdk-lite gives you:

• An agent graph — LLM nodes, function nodes, delegation nodes (agent-to-agent), aggregator nodes (fan-in), and planner nodes (dynamic DAGs), wired together with explicit routing strategies.
• A messaging layer — pydantic wire models (AgentTaskMessage, StreamFragment, delegation contracts) and a RabbitMQ transport (client, consumer, publisher, RPC, delegation transport) built on aio-pika and RabbitMQ Streams.
• A runner — AgentRunner consumes tasks from a queue, resolves per-request settings, executes your agent, and streams the response back — with graceful shutdown, retries, and circuit breakers built in.
• Bring your own LLM — LLMCallable is a protocol. Wrap any HTTP client, local model, or provider SDK in ~15 lines; no vendor lock-in.

Nothing here depends on a private backend, an internal config service, or a specific LLM vendor. Everything infrastructure-specific is a constructor argument or a protocol you implement yourself.

Install

pip install sofias-sdk-lite
# or
uv add sofias-sdk-lite

Requires Python 3.11+ and a running RabbitMQ broker (with the Streams plugin enabled) if you use the runner or streaming responses.

Quickstart

A minimal agent with a single function node:

import asyncio

from sofias_sdk_lite import (
    AgentBuilder,
    AgentMessage,
    BaseAgentSettings,
    InputContract,
    NodeContract,
    OutputContract,
)


class GreetInput(InputContract):
    name: str


class GreetOutput(OutputContract):
    greeting: str


class HelloSettings(BaseAgentSettings):
    pass


def greet(data: dict, context: dict | None = None) -> dict:
    return {"greeting": f"Hello, {data['name']}!"}


async def main() -> None:
    contract = NodeContract(input_schema=GreetInput, output_schema=GreetOutput)
    agent = (
        AgentBuilder("hello_agent", version="0.1.0")
        .with_settings_class(HelloSettings)
        .with_contract(input_schema=GreetInput, output_schema=GreetOutput)
        .add_function_node("greeter", contract, process_fn=greet)
        .set_entry_node("greeter")
        .set_terminal("greeter")
        .build()
    )

    response = await agent.execute(AgentMessage(content=GreetInput(name="World")))
    print(response.content)  # {"greeting": "Hello, World!"}


asyncio.run(main())

See examples/ for a runner against a local RabbitMQ (docker-compose.yml included), tool loops, streaming, and agent-to-agent delegation.

Running an agent against RabbitMQ

from sofias_sdk_lite import AgentRunner, RunnerConfig, RabbitMQConfig, AgentMessage


class MyAgentRunner(AgentRunner):
    settings_class = MySettings

    def build_agent(self, settings, workflow):
        return (
            AgentBuilder("my_agent")
            .with_settings_class(type(settings))
            .with_response_workflow(workflow)
            # ... nodes, routing ...
            .build()
        )

    def prepare_input(self, task, history, role):
        return AgentMessage(
            content=MyInput(message=task.content, role=role),
            conversation_id=task.conversation_id,
        )


if __name__ == "__main__":
    MyAgentRunner(
        RunnerConfig(
            queue="my-agent-tasks",
            agent_name="my_agent",
            rabbitmq=RabbitMQConfig(host="localhost"),
        )
    ).run()

Optional extras

• sofias-sdk-lite[otel] — installs opentelemetry-api so the runner opens an agent.handle span per turn and response fragments carry the active span's traceparent. Without it, the inbound traceparent is still echoed end to end.

Scope (v1)

Core agent graph, RabbitMQ messaging, and the runner ship today. Memory, MCP tool discovery, and LLM adapter implementations are intentionally out of scope for v1 — the SDK ships the relevant protocols (MemoryProvider, ToolProvider, LLMCallable) so you can plug in your own, and these become optional extras in a later release.

License

Apache-2.0. See LICENSE.