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()

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.