Skarta

---

Build agents in Python. Ship them on a Rust runtime.

Skarta is a production-grade runtime for multi-agent systems. You write the agents. Skarta handles orchestration, validation, scheduling, streaming, sessions, budgets, telemetry, and security.

pip install skarta

That single install ships a Python SDK and a pre-built runtime binary for macOS (Apple Silicon) and Linux (x86_64, aarch64). No Rust toolchain. No separate server.

Platform support

• Windows is not natively supported in this release. The runtime uses Unix domain sockets for local IPC, which require a Unix-family host. Windows users can run Skarta today via WSL2. Native Windows wheels are on the roadmap.
• macOS Intel (x86_64) is not built by CI. Apple Silicon Macs only for the macOS wheel today. Intel-Mac users can run the manual release script on an x86_64 macOS host or wait for native support.
• Sandboxed code execution is Linux / Docker only for now. The capability sandbox that isolates untrusted or model-generated code in a subprocess (the Sandbox allowlists) cannot yet launch an interpreter such as Python under the native macOS backend. Run such code on a Linux host (native sandbox) or via the sarthiai/skarta Docker image. Everything else, including runtime-enforced tool permissions and the network / filesystem / env allowlists, works on macOS as documented; only subprocess sandboxing of untrusted code is affected.

Client / server architecture

Beyond the single-install laptop flow, Skarta also runs as a standalone, network-exposed server that your code connects to remotely. The skarta-client PyPI wheel (client only, no bundled runtime) dials a running runtime over WebSocket or gRPC, with TLS, certificate pinning, server-issued scoped API keys, and SQLite or Postgres persistence. Self-host with the sarthiai/skarta Docker image. The full operator guide is docs/deployment.md.

---

See it work in one minute

A working two-agent typed pipeline that calls a real LLM. One terminal, one Python file, one env var.

1. Install:

pip install skarta
export OPENAI_API_KEY=sk-...

2. Save this as pipeline.py:

import asyncio
from pydantic import BaseModel
from skarta import LLM, Agent, Pipeline


class Topic(BaseModel):
    subject: str

class Brief(BaseModel):
    summary: str

class Headline(BaseModel):
    text: str


llm = LLM(model="gpt-4o-mini")

summarize = Agent(
    llm=llm,
    instructions="Write one sentence on the given subject.",
    input=Topic, output=Brief,
)
headline = Agent(
    llm=llm,
    instructions="Rewrite the brief as a punchy six-word headline.",
    input=Brief, output=Headline,
)


async def main() -> None:
    pipeline = Pipeline(summarize, headline)
    result = await pipeline.execute(Topic(subject="production-grade agent runtimes"))
    print(result.text)


if __name__ == "__main__":
    asyncio.run(main())

3. Run:

python pipeline.py

You will see a real LLM-generated headline.

What just happened: Skarta started, picked up your OpenAI key, saw that summarize produces a Brief and headline consumes a Brief, worked out the order itself, called OpenAI for the first agent, checked the reply matched the shape you asked for, fed it into the second agent, called OpenAI again, checked that reply too, and gave you back a finished Headline object. You wrote no flow chart, no retry loop, no queue, no glue.

Prefer Anthropic? Swap the LLM declaration:

llm = LLM(model="claude-haiku-4-5")  # picks up ANTHROPIC_API_KEY

OpenRouter, Groq, Ollama, or any OpenAI-compatible endpoint: pass base_url and provider="openai_compat" to LLM(...). For richer config (multiple providers, pricing overrides, persistent storage), drop a framework.toml next to your script. See Configuring an LLM endpoint for the full setup.

Want streaming, sessions, tools, or multi-agent collaboration? Three more lines:

from skarta import Tool, Memory, Team, Schedule

@Tool
async def search(args): ...                                  # tool the agent can call

agent = Agent(llm=llm, instructions="...", tools=[search])
agent = Agent(llm=llm, instructions="...", memory=Memory())  # cross-call memory
team  = Team(members=[a, b, c], shape="brainstorm")          # 3-agent dialogue

@Schedule("0 9 * * *")                                       # cron-trigger an agent
async def daily_brief(): await agent.execute("...")

The full surface is in Getting started and per-topic pages in docs/.

---

From your first agent to a production service

Eleven runnable files, each one a complete example. Each step adds one capability. Stop where Skarta does enough for what you're building. Every example assumes:

pip install skarta
export OPENAI_API_KEY=sk-...

Level 1. One agent that answers

Single-agent, single LLM call.

A customer types a question, your agent answers. The shortest possible Skarta program.

import asyncio
from skarta import LLM, Agent

agent = Agent(
    llm=LLM(model="gpt-4o-mini"),
    instructions="Answer in one short paragraph.",
)

async def main():
    print(await agent.execute("Why are Rust runtimes a good fit for agents?"))

asyncio.run(main())

Skarta started, called OpenAI, returned the answer, and shut down on its own. Nine lines, no server to run.

Level 2. Get the answer in the exact shape your code expects

Structured outputs with typed I/O.

The model returns a Python object with named fields, not a string of text you have to parse and pray over.

import asyncio
from pydantic import BaseModel
from skarta import LLM, Agent

class Question(BaseModel):
    text: str

class Answer(BaseModel):
    reply: str
    confidence: float

agent = Agent(
    llm=LLM(model="gpt-4o-mini"),
    instructions="Reply briefly. Include a confidence score from 0 to 1.",
    input=Question, output=Answer,
)

async def main():
    out = await agent.execute(Question(text="Is the sky blue?"))
    print(out.reply, out.confidence)

asyncio.run(main())

If the model returns junk or skips a field, Skarta raises a clear error before the bad value reaches your code. You can act on out.confidence and out.reply knowing they are exactly the types you asked for.

Level 3. Let the agent call your own functions

Tool calling (also called function calling), with the tool loop on autopilot.

Hand the agent a Python function that hits your API, queries your database, or returns a price from your pricing service. The agent decides when to call it.

import asyncio
from pydantic import BaseModel
from skarta import LLM, Agent, Tool

class WeatherArgs(BaseModel):
    city: str

@Tool
async def get_weather(args: WeatherArgs) -> dict:
    return {"city": args.city, "temp_c": 23, "condition": "sunny"}

agent = Agent(
    llm=LLM(model="gpt-4o-mini"),
    instructions="If asked about weather, call get_weather.",
    tools=[get_weather],
)

async def main():
    print(await agent.execute("What's the weather in Tokyo right now?"))

asyncio.run(main())

Skarta did the back-and-forth: the model asked to call get_weather, Skarta ran your function, fed the result back to the model, and the model wrote its final answer. None of that wiring was yours to write.

Level 4. An agent that remembers what the customer said last time

Conversational memory with pluggable session storage.

The agent picks up where the conversation left off. Same code works for a five-minute chat or one that spans a week.

import asyncio
from skarta import LLM, Agent, Memory

agent = Agent(
    llm=LLM(model="gpt-4o-mini"),
    instructions="You are a helpful note-taker.",
    memory=Memory(),
)

async def main():
    await agent.execute("My favourite colour is teal.", session="user-42")
    print(await agent.execute("What is my favourite colour?", session="user-42"))

asyncio.run(main())

Skarta loaded the earlier turns for session user-42, gave them to the model as context, and saved the new exchange back. Conversations live in a file by default. Switch to Memory(backend="db") and they ride along through restarts and across machines.

Level 5. Save a chat, try a different reply, jump back if you don't like it

Conversation branching and named session checkpoints.

Conversations aren't write-once. Save the chat at any turn, test a tangent, and roll back to the saved turn if the new direction was worse.

import asyncio
from skarta import LLM, Agent, Memory

mem = Memory()
agent = Agent(
    llm=LLM(model="gpt-4o-mini"),
    instructions="You are a trip-planning assistant.",
    memory=mem,
)

async def main():
    await agent.execute("Let's plan a 3-day trip to Kyoto.", session="trip")
    await mem.checkpoint("trip", name="after-outline")

    # Try a tangent on the same thread.
    await agent.execute("Actually, swap day 2 for a day-trip to Osaka.", session="trip")

    # Decide we preferred the original; rewind to the pinned point.
    await mem.restore("trip", name="after-outline")
    print(await agent.execute("OK keep Kyoto only. What's day 1?", session="trip"))

asyncio.run(main())

Skarta saved the conversation at the moment you named, let the agent try the Osaka tangent, then dropped that tangent and reset to the Kyoto plan. Use mem.fork(...) instead of restore to keep both versions alive and switch between them later.

Level 6. Hand the work along a line of agents

Agent pipeline with auto-DAG: Skarta derives the order from your data shapes.

A research question comes in. One agent plans the bullet points, the next writes a draft, the third compresses it to a one-page brief. You write the three agents; Skarta works out the order from your data and runs them back to back.

import asyncio
from pydantic import BaseModel
from skarta import LLM, Agent, Pipeline

class Question(BaseModel):
    text: str
class Outline(BaseModel):
    points: list[str]
class Draft(BaseModel):
    body: str
class OnePager(BaseModel):
    summary: str

llm = LLM(model="gpt-4o-mini")
planner   = Agent(llm=llm, instructions="List three bullet points to research.",
                  input=Question, output=Outline)
writer    = Agent(llm=llm, instructions="Write a short answer from these points.",
                  input=Outline,  output=Draft)
condenser = Agent(llm=llm, instructions="Compress the draft to about 80 words.",
                  input=Draft,    output=OnePager)

async def main():
    out = await Pipeline(planner, writer, condenser).execute(
        Question(text="How does a Rust agent runtime differ from a Python library?")
    )
    print(out.summary)

asyncio.run(main())

No flow charts, no edge lists. Skarta saw that planner produces an Outline, writer expects an Outline and produces a Draft, condenser expects a Draft, worked out the order, and che