actant 0.2.0

Temporal-native runtime for persistent agents with durable inboxes, governed tools, and replay

Actant

Actant is a durable Python agent runtime built on Temporal. Define agents and tools normally; Actant handles parallel tools, human approval, deferred work, subagents, suspension, and crash-safe continuation.

Actant is pre-1.0. Public APIs may change.

Why Actant

Agent tools become a distributed-systems problem when calls run in parallel, wait for people, or outlive a worker. Actant handles that orchestration:

• allowed tools execute concurrently;
• deferred tools pause without holding a worker;
• the next model turn waits for the complete tool group;
• approvals and nested-agent waits surface through the same API;
• Temporal recovers execution after process or worker failure;
• projection stores keep state easy for APIs and UIs to read.

flowchart TB
    Turn["One agent turn emits A, B, and C"]
    A["A: execute → completed"]
    B["B: wait ··· human approves → completed"]
    C["C: execute → completed"]
    Barrier["Durable tool-group barrier"]
    Next["Next agent turn"]

    Turn --> A & B & C
    A --> Barrier
    B --> Barrier
    C --> Barrier
    Barrier --> Next

Read Why Actant? for the detailed guarantees and framework comparison.

Install

pip install actant
pip install "actant[openai]"     # optional provider
pip install "actant[anthropic]"  # optional provider
pip install "actant[gemini]"     # optional provider

Start a local Temporal development server:

actant server start --detach

Quickstart

This complete example streams tokens and then prints the persisted final response:

import asyncio
from contextlib import suppress
from uuid import uuid4

from actant import AgentDefinition
from actant.llm.messages import Message
from actant.llm.providers.fake import FakeLLM, FakeResponse
from actant.runtime import AgentRuntime, TemporalRuntimeConfig, TemporalRuntimeWorker
from actant.runtime.events import AgentThreadHooks, StreamListener
from actant.runtime.stores import InMemoryRuntimeStores
from actant.tools import ToolRegistry

responses: asyncio.Queue[Message | Exception] = asyncio.Queue()
stores = InMemoryRuntimeStores()
config = TemporalRuntimeConfig(address="localhost:7233")


class Hooks(AgentThreadHooks):
    async def on_assistant_message(self, message: Message) -> None:
        if not message.tool_calls:
            await responses.put(message)

    async def on_error(self, error: Exception) -> None:
        await responses.put(error)


class Stream(StreamListener):
    async def on_text_delta(self, delta: str) -> None:
        print(delta, end="", flush=True)


agent = AgentDefinition(
    id="assistant",
    name="Assistant",
    persona="You are a useful assistant.",
    llm=FakeLLM(
        [
            FakeResponse(
                text="Hello from Actant.",
                text_chunks=["Hello ", "from ", "Actant."],
            )
        ]
    ),
    tools=ToolRegistry([]),
)
agents = {agent.id: agent}

runtime = AgentRuntime(stores=stores, agents=agents, temporal=config)
worker = TemporalRuntimeWorker(
    stores=stores,
    agents=agents,
    config=config,
    hooks_factory=lambda _thread: Hooks(),
    listener_factory=lambda _thread: Stream(),
)


async def main() -> None:
    worker_task = asyncio.create_task(worker.run())
    try:
        print("Streaming: ", end="", flush=True)
        await runtime.send_message(agent.id, uuid4().hex, "hello")
        response = await asyncio.wait_for(responses.get(), timeout=60)
        if isinstance(response, Exception):
            raise response
        print(f"\nFinal: {response.content}")
    finally:
        worker_task.cancel()
        with suppress(asyncio.CancelledError):
            await worker_task


asyncio.run(main())

# Streaming: Hello from Actant.
# Final: Hello from Actant.

send_message() durably submits work and returns immediately. StreamListener receives live deltas, hooks receive persisted lifecycle events, and the message store provides durable reload.

Use OpenAIProvider, AnthropicProvider, GeminiProvider, or QwenProvider in place of FakeLLM. Actant never chooses a model ID for you.

Demo

The included FastAPI + React viewer demonstrates streaming, approvals, multiple-choice questions, mixed parallel tools, and nested subagents without an API key:

just demo-sync
just demo

Open http://localhost:5173.

Documentation

• Core concepts
• Runtime architecture
• Runtime and deployment
• Tools and admission
• Pauses and deferred work
• Subagents
• Application coordinators
• Release process

Development

just sync
just test
just lint
just typecheck
just package

The justfile is repository-only. Installed users receive the actant CLI; run actant server --help for local Temporal commands.

License

MIT