Ai.os

Persistent agent runtime. Deploy agents that remember, survive crashes, and run forever.

from aios import Agent, tool

class ResearchAgent(Agent):
    name = "researcher"
    model = "claude-sonnet-4-6"

    @tool
    async def search_web(self, query: str) -> str:
        """Search the web. query: What to search for."""
        ...

    async def run(self):
        results = await self.search_web("latest AI papers")
        await self.memory.save("results", results)  # persists across restarts

if __name__ == "__main__":
    ResearchAgent.launch()

aios run researcher.py --detach   # run in background
aios logs researcher -f           # stream live logs
aios memory researcher            # inspect what it remembers
aios restart researcher           # restart — picks up where it left off
aios ui                           # open web dashboard

Why Ai.os

Every agent framework makes you solve the same problems from scratch:

Install

pip install aios-runtime

With Postgres support:

pip install "aios-runtime[postgres]"

Requires Python 3.10+. No Docker, no Redis, no external services.

Quickstart

aios init myagent          # scaffold a new agent project
cd myagent
echo "ANTHROPIC_API_KEY=your-key" >> .env
aios run myagent.py        # run it

Core API

Agent base class

from aios import Agent, tool, schedule

class MyAgent(Agent):
    name = "myagent"             # persistent identity key
    model = "claude-sonnet-4-6"  # or "gpt-4o", "ollama/llama3"
    version = "1.0.0"
    description = "What this agent does"
    system_prompt = "You are..."
    temperature = 0.7
    config = {"key": "value"}    # arbitrary config, persisted

    @tool
    async def my_tool(self, input: str) -> str:
        """Tool description shown to the LLM. input: What it means."""
        return f"result: {input}"

    async def run(self) -> None:
        ...

if __name__ == "__main__":
    MyAgent.launch()

Memory

# Short-term (cleared each run)
self.memory.set("draft", "...")
value = self.memory.get("draft", default=None)
self.memory.clear()

# Long-term (persisted forever across runs)
await self.memory.save("report", {"rows": 42})
result = await self.memory.load("report", default=None)
await self.memory.delete("report")
keys = await self.memory.keys()
all_data = await self.memory.all()

# Search (case-insensitive substring match on keys and values)
results = await self.memory.search("climate", limit=5)
# → [{"key": "finding:climate", "value": {...}, "updated_at": "..."}]

# Timeline (append-only event log)
await self.memory.log_event("analysis_complete", {"rows": 1200})
events = await self.memory.timeline(limit=100)

Crash recovery

Before each @tool call, the result is cached in SQLite. If the agent crashes and restarts, run() re-executes from the top — but every completed tool call returns its cached result instantly. The agent fast-forwards to the first uncompleted call and continues.

No manual checkpoints. No configuration. It just works.

Webhook triggers

Make an agent respond to external HTTP events instead of running once:

from aios import Agent, trigger

class GitHubReviewAgent(Agent):
    name = "gh_reviewer"
    model = "claude-sonnet-4-6"

    @trigger("webhook", path="/github", port=8080, secret="env:GITHUB_WEBHOOK_SECRET")
    async def run(self, payload: dict) -> None:
        pr = payload.get("pull_request", {})
        if not pr:
            return
        review = await self.think(f"Review this PR: {pr['title']}")
        await self.memory.save(f"review_{pr['number']}", review)

if __name__ == "__main__":
    GitHubReviewAgent.launch()

Point any webhook source at http://your-host:8080/github. Works with GitHub, Stripe, Slack slash commands, or any HTTP sender. HMAC-SHA256 signature verification built in.

Tool retries

@tool(retries=3, backoff=2.0)
async def fetch_data(self, url: str) -> str:
    """Fetch from an external API. url: endpoint to call."""
    # Automatically retried up to 3 times on exception.
    # Wait: 2s → 4s → 8s (exponential backoff).
    return await self.http_get(url)

Tool result caching

@tool(cache_ttl=60)
async def get_price(self, symbol: str) -> str:
    """Get current price. symbol: ticker to look up."""
    # Same args → same result returned instantly for 60 seconds.
    # Saves API calls and speeds up repeated LLM loops.
    return await self.http_get(f"https://api.example.com/price/{symbol}")

Combine with retries: @tool(retries=3, backoff=2.0, cache_ttl=300)

Scheduling

class DailyAgent(Agent):
    @schedule("every 24h")       # or "every 30m", "every 1d"
    async def run(self):
        await self.do_daily_work()

Next-run time persists in memory — correct schedule survives restarts.

LLM calls

# Single-shot, no tools
text = await self.think("Summarize this: ...")

# Agentic loop — LLM selects and calls tools until done
answer = await self.think_with_tools(
    "Research X, save findings, return a summary",
    max_iterations=10,
)

Agent-to-agent

# Blocking — call another agent, get text back
summary = await self.call_agent(ResearchAgent, "summarize the findings on X")

# Fire-and-forget — spawn in background
await self.spawn_agent(MonitorAgent)

Lifecycle hooks

async def on_start(self) -> None:
    """Called once before run(). Setup work goes here."""

async def on_stop(self) -> None:
    """Called after run() completes or crashes. Cleanup goes here."""

Logging

Every agent has a self.logger (a standard logging.Logger) pre-configured with the agent's name:

async def run(self) -> None:
    self.logger.info("Starting run")
    self.logger.debug("Fetching %s", url)
    self.logger.warning("Rate limit hit, backing off")

Logs appear in aios logs <name> and the web UI log viewer.

Multi-model

Change one line — no other code changes:

model = "claude-sonnet-4-6"      # Anthropic
model = "claude-opus-4-8"        # Anthropic (most capable)
model = "gpt-4o"                 # OpenAI
model = "gpt-4o-mini"            # OpenAI (fast + cheap)
model = "gemini/gemini-pro"      # Google
model = "mistral/mistral-large"  # Mistral
model = "ollama/llama3"          # Local — no API key needed
model = "ollama/mistral"         # Local Mistral

Powered by litellm — any OpenAI-compatible endpoint works.

CLI reference

Environment variables

Copy .env.example to .env:

cp .env.example .env

ANTHROPIC_API_KEY=...        # claude-* models
OPENAI_API_KEY=...           # gpt-* models
GOOGLE_API_KEY=...           # gemini-* models
AIOS_LOG_LEVEL=INFO          # DEBUG | INFO | WARNING | ERROR
AIOS_ALERT_WEBHOOK=https://… # POST here when an agent crashes (Slack/Discord/custom)

Ai.os auto-loads .env on Agent.launch(). Local Ollama needs no key.

Encrypted secrets (production)

For production deployments where you don't want plain-text .env files, use the built-in encrypted secrets store:

pip install cryptography          # one-time install

aios secrets set OPENAI_API_KEY sk-...
aios secrets set SLACK_BOT_TOKEN xoxb-...
aios secrets import .env          # or bulk-import an existing .env file
aios secrets list                 # see stored names (never values)

Secrets are encrypted with Fernet symmetric encryption. The master key lives at ~/.aios/master.key (chmod 600 on Unix) and encrypted values at ~/.aios/secrets.db.

Every agent automatically calls SecretsStore.inject_to_env() at startup — secrets are merged into os.environ (existing variables are never overwritten). Agents require zero code changes; just stop shipping the .env file and store secrets once with aios secrets set.

Scrape Prometheus metrics from the web dashboard:

http://localhost:8000/metrics

Built-in tool mixins

Mix in capabilities by inheriting alongside Agent:

from aios import Agent, WebSearchMixin, FilesystemMixin, SlackMixin, PostgresMixin

class AnalystAgent(Agent, WebSearchMixin, FilesystemMixin, SlackMixin, PostgresMixin):
    name = "analyst"
    model = "claude-sonnet-4-6"

    async def run(self):
        rows = await self.pg_query("SELECT * FROM orders WHERE status = $1", ["pending"])
        summary = await self.think(f"Summarize these {len(rows)} orders: {rows[:5]}")
        await self.slack_send_message("#data-team", summary)
        await self.write_file("report.md", summary)

Examples

TypeScript / Node.js SDK

Same API, same persistence layer:

import { Agent, tool, schedule } from '@aios/sdk';

class ResearchAgent extends Agent {
  static name = 'researcher';
  static model = 'claude-sonnet-4-6';

  @tool({ description: 'Save a finding' })
  async saveFinding(topic: string, summary: string): Promise<string> {
    this.memory.save(`finding:${topic}`, { summary });
    return `Saved: ${topic}`;
  }

  async run(): Promise<void> {
    const result = await this.thinkWithTools('Research AI agents and save findings.');
    this.memory.save('last_run', result);
  }
}

ResearchAgent.launch();

cd sdk/typescript
npm install
npx ts-node examples/researcher.ts

A Python agent and a TypeScript agent with the same name share one SQLite database — memory persists across both runtimes.

Team Workspaces

Share agent memory across machines without a server:

# On machine A (set up once)
aios workspace init myteam --dir /mnt/shared/aios   # or ~/Dropbox/aios

# Push agent state to the shared dir
aios workspace push researcher

# On machine B
aios workspace init myteam --dir /mnt/shared/aios
aios workspace pull researcher                       # merge
aios workspace pull researcher --replace             # overwrite
aios workspace list                                  # see what's available

Works with any shared filesystem — NFS, Dropbox, Google Drive, S3 mounted via s3fs, etc.

Encrypted Secrets (Production)

Stop putting API keys in plain-text .env files:

aios secrets set ANTHROPIC_API_KEY sk-ant-...
aios secrets set SLACK_BOT_TOKEN xoxb-...
aios secrets list                              # shows names only, never values
aios secrets import .env                       # bulk import from existing .env
aios secrets get ANTHROPIC_API_KEY             # decrypt + print

Secrets are Fernet-encrypted in ~/.aios/secrets.db. Agents automatically receive them at startup — no code changes needed.

pip install "aios-runtime[secrets]"   # or: pip install cryptography

Remote Log Streaming

Watch a cloud-hosted agent's logs from your laptop:

# Stream logs from a remote Ai.os web UI
aios logs myagent --remote ws://my-server:8000

# Or connect directly with any WebSocket client
wscat -c ws://my-server:8000/ws/agents/myagent/logs

vs. alternatives

Roadmap

v0.1 — Foundation ✅

• Agent persistence + crash recovery
• Short-term + long-term memory
• @tool decorator with auto schema
• Multi-model routing (Claude, GPT-4o, Gemini, Ollama, any OpenAI-compatible)
• CLI (run / list / logs / stop / restart / memory / doctor)
• Web UI dashboard with live log streaming
• @schedule decorator with persistent next-run
• Agent-to-agent calls (call_agent / spawn_agent)
• aios init project scaffolding
• Built-in tool library (Web search, Filesystem, Shell, HTTP, GitHub)

v0.2 — SDKs & Ecosystem ✅

• Slack, Discord, Email tool mixins
• Postgres, Notion, Linear tool mixins
• aios export / aios import — memory backup and migration
• Web UI — dark/light theme, relative timestamps, ANSI stripping, noise filter, Trace tab
• TypeScript / Node.js SDK — same API, same SQLite persistence layer (sdk/typescript/)
• aios publish — share agents as pip-installable packages

v0.3 — Observability ✅

• Timeline tab in web UI
• Tool call Trace tab (see exactly which tools were cached for crash recovery)
• Prometheus metrics export — /metrics endpoint, scrape with Grafana / Datadog
• Alert webhooks on agent failure — AIOS_ALERT_WEBHOOK=<url>

v0.4 — Cloud ✅

• aios deploy — generate production deploy bundle (Docker Compose, Fly.io, systemd)
• @tool(retries=N, backoff=s) — automatic exponential-backoff retry on tool failure
• aios publish — scaffold and upload agents as pip-installable packages
• @trigger("webhook") — event-driven agents; respond to GitHub, Stripe, Slack, etc.
• Team workspaces — aios workspace push/pull over any shared directory
• Secrets management — Fernet-encrypted secrets store (aios secrets)
• Remote log streaming — WebSocket endpoint + aios logs --remote

v0.5 — Visual Builder ✅

• Drag-and-drop workflow editor in web UI (Workflow tab — nodes, edges, canvas)
• Export workflow to Python agent class (Export .py button)
• CSV / JSON / PDF exports from web UI (Runs, Memory, Timeline, full report)

Contributing

See CONTRIBUTING.md. PRs welcome.

License

MIT — see LICENSE.