Python Agent SDK

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):
    class="token comment"># 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

class="token comment"># Agent workspace is a Chronicle FUSE mount
workspace = Path(os.environ['AGENT_WORKSPACE'])

class="token comment"># All file operations are automatically versioned
(workspace / 'output.txt').write_text('Hello, world!')

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