​
Quick Start

from podium.agent import BaseAgent, AgentContext
from podium.agent.models.delta_state import DeltaState


class MyAgent(BaseAgent):
    class State(DeltaState):
        _version = 1
        _entity_type = "my_agent"
        count: int = 0
        history: list[str] = []

    async def on_message(self, ctx: AgentContext, message: dict) -> None:
        ctx.state.count += 1
        ctx.state.history.append(message.get("text", ""))

        msg = await ctx.send_message("")
        await msg.stream_token(
            f"Hello! You've sent {ctx.state.count} messages.", final=True
        )

​
Agent Lifecycle

​
BaseAgent

class MyAgent(BaseAgent):
    # State class for auto-persistence (optional)
    class State(DeltaState):
        _version = 1
        _entity_type = "my_agent"

    async def on_init(self, ctx: AgentContext) -> None:
        """Called once after runtime is initialized."""
        pass

    async def on_message(self, ctx: AgentContext, message: dict) -> None:
        """Handle inbound message. All output via streaming."""
        raise NotImplementedError

    async def on_shutdown(self, ctx: AgentContext) -> None:
        """Called during graceful shutdown."""
        pass

    async def on_idle_timeout(self, ctx: AgentContext) -> bool:
        """Called before scale-to-zero. Return True to allow."""
        return True

​
DeltaState

class ConversationState(DeltaState):
    _version = 2
    _entity_type = "conversation"

    turn_count: int = 0
    messages: list[dict] = []
    settings: dict = {"model": "claude-sonnet-4", "temperature": 0.7}
State mutations are tracked automatically:
  • Simple attribute changes generate replace deltas
  • List modifications generate add/remove deltas
  • Nested dict changes are tracked recursively

​
Runtime

Python agents run inside a Rust shim that embeds the Python interpreter via PyO3:
  • The Rust shim handles IPC communication with the coordinator
  • Python GIL is managed by the Rust layer
  • Agent code runs in an asyncio event loop

​
Workspace Access

import os
from pathlib import Path

# Agent workspace is a Chronicle FUSE mount
workspace = Path(os.environ['AGENT_WORKSPACE'])

# All file operations are automatically versioned
(workspace / 'output.txt').write_text('Hello, world!')

# Query version history via magic file API
history = (workspace / '.chronicle' / 'query' / 'output.txt').read_text()