atomr-agents-ai-skills

atomr-agents

A native Rust agentic framework built as a layered actor / strategy / harness substrate on top of atomr and atomr-infer. atomr-agents gives you a single mental model — pluggable strategies that resolve under shared budgets, channelled state with first-class checkpointing, tool-call orchestration with parallel dispatch, and durable harness loops — that scales from a one-off chatbot to a multi-tenant production agent platform.

use atomr_agents::prelude::*;

// One Pipeline composes prompt → model → parser like LCEL.
let pipeline = Pipeline::from(prompt)
    .then(model)
    .then(parser)
    .build();

let answer = pipeline.call(input, ctx).await?;

Python parity

The Python facade ships every Rust capability. The native extension atomr_agents._native is split into 28 hierarchical submodules: foundational (errors, core, callable_, strategy, instruction, context, state, observability, cache, parser, registry); tool / skill / memory / retrieval / ingest (tool, skill, memory, embed, retriever, ingest, persona); agent / workflow / harness / org / eval (agent, workflow, harness, org, eval); voice (stt, tts, voice, voice_extras); plus the guest registry. The top-level package re-exports the most-used classes, ships a PEP 561 py.typed marker, and exposes async coroutines / async iterators over pyo3-async-runtimes.

Install

pip install atomr-agents

For an editable workflow against the local checkout:

pip install maturin
maturin develop --features python -m crates/py-bindings/Cargo.toml
pip install -e ".[dev]"

Host-mode async event stream

EventBus.stream() returns an EventStream that implements the Python async iterator protocol. Drive a producer on the same loop and consume events as they fire:

import asyncio
from atomr_agents.observability import EventBus


async def main() -> None:
    bus = EventBus()
    stream = bus.stream()

    bus.emit_tool_invoked("calc", args_hash=0, elapsed_ms=5, ok=True)
    bus.emit_tool_invoked("search", args_hash=1, elapsed_ms=12, ok=True)

    async for ev in stream:
        print(ev.kind, ev.timestamp_ms)
        if ev.kind == "tool_invoked" and ev.tool == "search":
            break


asyncio.run(main())

Async registry publish

Registry.publish_async returns a Python awaitable backed by a tokio future, so version pins land without blocking the event loop:

import asyncio
from atomr_agents.registry import Registry


async def main() -> None:
    registry = Registry()
    record = await registry.publish_async(
        "tool_set", "calc", "0.1.0", {"name": "calc"}
    )
    print(record.kind, record.id, record.version)


asyncio.run(main())

Guest-mode @tool decorator

atomr_agents.guest exposes real decorators wired through _native.guest.register_*_factory. A guest tool is a class with an async def invoke(self, args, ctx) method:

from atomr_agents.guest import tool


@tool(toolset="calc")
class Add:
    name = "add"

    async def invoke(self, args: dict, ctx) -> dict:
        return {"sum": args["a"] + args["b"]}

Mirror decorators are available for the full set of 24 Rust traits: @strategy(kind=...), @persona, @skill, @parser, @scorer, @memory_store, @embedder, @callable_, @retriever, @loader, @splitter, @kv_cache, @long_store, @tracer, @conversation_agent, @diarizer, @vad, @phonemizer, @journal, @repair_model, @persona_reconciler, @inference_client, @ann_index. Each pairs with an atomr_agents.<module>.*_from_factory(key) builder that materialises the registered Python target as a Rust dyn handle.

Host-mode agent runtime

AgentBuilder assembles strategy slots into a runnable AgentRef that implements Callable, so an agent composes with the same decorators and pipeline operators as any other unit:

from atomr_agents.agent import AgentBuilder
from atomr_agents.harness import Harness, iteration_cap, loop_strategy_from_callable
from atomr_agents.workflow import Dag, Step, WorkflowRunner

# Strategy slots come from in-process factories or Python guests.
builder = AgentBuilder("research-agent", "gpt-4o-mini")
builder.with_instructions(instructions)
builder.with_tools(tool_strategy)
builder.with_memory(memory_strategy)
builder.with_skills(skill_strategy)
builder.with_inference(inference_client)
agent_ref = builder.build()
result = await agent_ref.run_turn("What's the GDP of France?")

# The agent is itself a Callable — drop it into a workflow.
dag = Dag("plan")
dag.add_step("plan", Step.invoke(agent_ref.as_callable()))
runner = WorkflowRunner("research-wf", dag.build())
await runner.run({"user": "..."})

Where things live