tvastar 0.4.0

Tvastar — a programmable agent harness framework for Python. Agent = Model + Harness.

Tvastar

Not another SDK. A lightweight Python harness with just enough framework to be useful — and knows when to stay out of the way.

Define your agent once. Run it safely anywhere. Tvastar handles safe code execution, crash recovery, silent failure detection, and deploy-anywhere portability — without making you learn a new way to think about agents.

pip install tvastar

---

What is a harness?

Most Python agent libraries give you one of two things: orchestration patterns (how agents coordinate) or model wrappers (how to call an LLM). Neither solves the problem of running agents safely in production.

A harness is the missing layer. It sits between your agent logic and the real world and handles what happens when things go wrong — code that crashes, context that overflows, silent failures, infrastructure that varies across environments.

Tvastar includes lightweight framework primitives so you have something to run (AgentSpec, @tool, sessions, workflows). But the framework is minimal on purpose. The harness is the product.

---

The four problems Tvastar solves

1. Running agent-produced code safely

Most frameworks assume you have a container. Tvastar runs real code in-memory with no Docker, no setup, no external service. Switch to Docker or a remote sandbox with one line when you need stronger isolation.

2. Agents that lie about success

An agent says "all tests pass" over a failing run. An agent claims a file was created but nothing was written. Tvastar detects silent failures automatically and surfaces them before they reach your users.

3. Long-running agents that crash

A 10-minute agent run failing at minute 9 loses everything. Tvastar checkpoints transcript and filesystem after every step. Crashes resume from where they stopped, not from the beginning.

4. Deploying the same agent everywhere

One agent definition runs as a web service, AWS Lambda, GitHub Action, container, or serverless function. No rewriting. No framework-specific deployment config.

---

Works with any agent or model

import asyncio
from tvastar import create_agent, Harness, default_toolset
from tvastar.model import AnthropicModel

# Wrapping a raw model call
agent = create_agent(
    "assistant",
    model=AnthropicModel("claude-opus-4-6"),
    instructions="You are a helpful coding agent.",
    tools=default_toolset(),
)
result = asyncio.run(Harness(agent).run("Write hello.py and run it."))
print(result.text)

# Wrapping an OpenAI-compatible provider
from tvastar.model import OpenAIModel

agent = create_agent("assistant", model=OpenAIModel("gpt-4o"), tools=default_toolset())

# Local Ollama — completely free, no API key
model = OpenAIModel(model="llama3.2", base_url="http://localhost:11434/v1", api_key="ollama")
agent = create_agent("assistant", model=model, tools=default_toolset())

# Any OpenAI-compatible provider (Groq, Together, Cloudflare…)
model = OpenAIModel(
    model="llama-3.1-8b-instant",
    base_url="https://api.groq.com/openai/v1",
    api_key="gsk_...",
)

The harness wraps the model. It does not care which one.

---

See it in action: tvastar-fix

The fastest way to understand Tvastar is to watch it fix something real.

tvastar-fix is a CLI tool and GitHub Action that auto-fixes failing tests. Your tests fail on a PR. Tvastar runs the agent, executes the fixes in a safe sandbox, verifies they actually pass, and pushes the correction — without you touching a line.

It is the reference implementation for everything the harness provides: safe execution, silent failure detection, crash recovery, and deploy-anywhere portability in one working example.

pip install "tvastar[fix]"
tvastar-fix --test-cmd "pytest tests/" --model claude-opus-4-6

---

When not to use Tvastar

• You only need a single chat completion → call the model SDK directly, Tvastar is overkill
• You need hundreds of pre-built integrations (Slack, Salesforce, databases) → LangChain's ecosystem is larger
• Your agent never executes code or writes files → the sandbox and failure detection add weight without benefit

Tvastar is for agents that do things — run code, edit files, call tools — and need to do those things safely in production. If your agent only talks, you do not need a harness.

---

What Tvastar handles so you do not have to

Problem	How Tvastar handles it
Code execution without Docker	In-memory sandbox, zero setup
Agent claims success but fails	Built-in silent failure detection
Crash at step 47 of 50	Step-level checkpoint and resume
Deploy to Lambda, GitHub Actions, web	Single agent definition, any target
Agent loops on the same tool	Built-in loop detection
Context grows past model limit	Automatic compaction and summarisation
Audit what the agent actually did	Full transcript stored every run
Flaky network tools fail mid-run	Per-tool retry with exponential backoff
Run 100 prompts at once	Built-in parallel fan-out
Stream tokens to the browser	SSE endpoint out of the box

---

How it works

create_agent(...)  →  AgentSpec          (what the agent is — immutable)
Harness(spec)      →  Harness            (how it runs — stateful)
harness.run(...)   →  RunResult          (one prompt, one answer)
harness.session()  →  Session            (multi-turn conversation)

Inside every run() or prompt(), the loop looks like this:

User message
    ↓
Model generates response
    ↓
  ┌─ stop_reason == TOOL_USE? ──────────────────────────────────┐
  │                                                             │
  │   Execute all requested tools (concurrently)               │
  │   Feed results back to model                               │
  │   Auto-compact context if policy threshold hit             │
  │   Checkpoint to durable store                              │
  │   Loop ────────────────────────────────────────────────────┘
  │
  └─ END_TURN → RunResult(.text, .messages, .usage, .steps, .data)

---

Install

pip install tvastar                      # core only — zero deps
pip install "tvastar[anthropic]"         # + Claude models
pip install "tvastar[openai]"            # + OpenAI / Groq / Ollama / etc.
pip install "tvastar[serve]"             # + HTTP server (FastAPI)
pip install "tvastar[otel]"              # + OpenTelemetry tracing
pip install "tvastar[all]"              # everything

---

Core concepts

Thing	What it is
AgentSpec	Immutable declaration: model + tools + instructions + policies
Harness	Stateful runtime: runs an AgentSpec across sessions
Session	One conversation thread with its own message history
Tool	A Python function the model can call (schema auto-derived)
Skill	A Markdown file of reusable expertise, loaded on demand
Sandbox	Where code runs — virtual (in-memory), local, or Docker
RunResult	What you get back: .text, .data, .usage, .steps, .ok

---

Tools

from tvastar import tool, ToolRetryPolicy

@tool
def add(a: int, b: int) -> int:
    "Add two integers."
    return a + b

# With retry for flaky network calls
@tool(retry=ToolRetryPolicy(max_attempts=3, backoff_base=0.5))
async def call_api(url: str) -> str:
    "Fetch a URL."
    ...

# Access session context (sandbox, filesystem, memory)
@tool
async def save(path: str, content: str, ctx: ToolContext) -> str:
    "Save a file."
    ctx.filesystem.write(path, content)
    return "saved"

Built-in tools via default_toolset(): bash, read_file, write_file, edit_file, grep, glob, list_files.

Harness-wide retry — applies to all tools that do not have their own policy:

agent = create_agent(..., tool_retry=ToolRetryPolicy(max_attempts=3))

---

Sessions

harness = Harness(agent)

# One-shot
result = await harness.run("Summarise this document.")

# Multi-turn
sess = harness.session()
async with sess:
    await sess.prompt("Read report.txt")
    await sess.prompt("Write a 3-bullet summary")
    result = await sess.prompt("Translate the summary to Spanish")

# Named sessions for parallel branches
branch_a = harness.session("review-api")
branch_b = harness.session("review-auth")
results = await asyncio.gather(
    branch_a.prompt("Review the API layer"),
    branch_b.prompt("Review the auth layer"),
)

---

Structured output

Get back a typed object instead of raw text:

from pydantic import BaseModel

class Report(BaseModel):
    summary: str
    issues: list[str]
    severity: str

result = await sess.prompt("Analyse this code.", result=Report)
report: Report = result.data
print(report.severity)

Works with Pydantic v2, Pydantic v1, dataclasses, plain dict, or any callable validator.

---

Delegating to specialist sub-agents

from tvastar import create_agent, define_agent_profile

reviewer = define_agent_profile(
    name="reviewer",
    description="Reviews code for security and correctness.",
    instructions="Report only issues with a reproducible failure scenario.",
    thinking_level="high",
    max_steps=10,
)

agent = create_agent("coordinator", model=model, subagents=[reviewer], tools=default_toolset())

sess = harness.session()
async with sess:
    result = await sess.task(
        "Review the auth package for security issues.",
        agent="reviewer",
        cancel_after=60.0,
        result=ReviewReport,
    )

Task delegation is capped at 4 levels deep to prevent runaway recursion.

---

Parallel fan-out

Run multiple prompts concurrently with one call:

results = await harness.fan_out([
    "Summarise chapter 1",
    "Summarise chapter 2",
    {
        "prompt": "Summarise chapter 3",
        "agent": "summariser",
        "cancel_after": 30.0,
        "result": SummarySchema,
    },
], concurrency=4)

---

Extended thinking

agent = create_agent(..., thinking_level="high")
# Anthropic: budget_tokens=16000  (low=1024, medium=8000, high=16000)
# OpenAI:    reasoning_effort='high'

---

Workflows — durable, inspectable pipelines

from tvastar import workflow
from tvastar.workflow import WorkflowContext

@workflow
async def summarise_document(ctx: WorkflowContext) -> dict:
    harness = await ctx.init(agent)
    sess = await harness.session()
    result = await sess.prompt(f"Summarise {ctx.payload['path']}")
    return {"summary": result.text, "steps": result.steps}

run = await summarise_document.run({"path": "report.pdf"})
print(run.status)   # RunStatus.COMPLETED
print(run.output)   # {'summary': '...', 'steps': 3}

for past_run in summarise_document.list_runs():
    print(past_run.run_id, past_run.status)

---

Event-driven dispatch

For chat bots, webhooks, and queue processors:

from tvastar import dispatch, dispatch_and_wait, observe_dispatch, DispatchInput

# Fire and forget
dispatch_id = await dispatch(
    agent,
    id="user_123",
    input=DispatchInput(text=message_text, type="chat.message"),
    on_complete=lambda r: send_reply(r.text),
    cancel_after=30.0,
)

# Fire and await
result = await dispatch_and_wait(agent, id="job_456", text="Process this report.")

# Watch all dispatches globally
observe_dispatch(lambda event: logger.info(event.type, extra=event.data))

---

Context compaction

Prevent context window exhaustion in long sessions:

from tvastar import CompactionPolicy

agent = create_agent(
    "long-runner",
    model=model,
    compaction=CompactionPolicy(
        max_messages=40,
        keep_last=10,
        min_messages=20,
    ),
)
# Fires automatically after tool turns. The model never notices.

---

Application-level file access

async with Harness(agent) as h:
    await h.fs.write_file("report.pdf", pdf_bytes)
    result = await h.run("Summarise report.pdf")
    summary = await h.fs.read_file("summary.md")

---

Sandboxes

from tvastar import VirtualSandbox, LocalSandbox, SecurityPolicy

# Default — in-memory, zero deps
create_agent(..., sandbox=VirtualSandbox)

# Real bash, jailed to a directory
policy = SecurityPolicy(allowed_commands={"python", "pytest"}, network=False)
create_agent(..., sandbox=lambda: LocalSandbox("./workspace", policy=policy))

---

MCP — use any published tool server

from tvastar import connect_mcp_server, default_toolset

client = await connect_mcp_server(command="python", args=["my_mcp_server.py"])
# or remote:
client = await connect_mcp_server(url="https://api.example.com/mcp", headers={...})

agent = create_agent("a", model=model, tools=[*default_toolset(), *client.tools])
await client.close()

---

Durable execution — survive crashes

from tvastar import Harness, FileStore

harness = Harness(agent, store=FileStore(".tvastar-state"))

# On restart — resume from last checkpoint
sess = harness.resume("sess_abc123") or harness.session()

---

Serving over HTTP

pip install "tvastar[serve]"
tvastar serve my_agent.py:agent --port 8000

Method	Path	Description
GET	/	Agent info
POST	/sessions	Create session
POST	/sessions/{id}/prompt	Send a message
WS	/sessions/{id}/stream	WebSocket streaming
GET	/sessions/{id}/stream?text=...	SSE streaming

curl -N "http://localhost:8000/sessions/sess_abc/stream?text=Hello"
# data: {"type": "text_delta", "data": {"text": "Hello"}}
# data: [DONE]

---

Observability

from tvastar import Tracer, ConsoleExporter, JSONLExporter

harness = Harness(agent, tracer=Tracer([
    ConsoleExporter(),
    JSONLExporter("trace.jsonl"),
]))

OpenTelemetry (Braintrust, Honeycomb, Datadog, Sentry):

pip install "tvastar[otel]"

from tvastar import OTelExporter
harness = Harness(agent, tracer=Tracer([OTelExporter()]))

---

Silent-failure detection

result = await harness.run("Fix all test failures.")

if not result.ok:
    for finding in result.warnings:
        print(f"[{finding.severity}] {finding.detector}: {finding.message}")
# → [WARNING] unverified_completion: model claimed success but last tool result shows failures

Built-in detectors: unknown_tool, schema_mismatch, thrash_loop, ignored_tool_error, unverified_completion, empty_answer, step_limit.

Write your own:

from tvastar.detect import Finding, Severity

def slow_run(ctx):
    if ctx.stopped == "max_steps":
        return [Finding("slow_run", Severity.WARNING, "hit the step ceiling")]
    return []

create_agent(..., detect=[*default_detectors(), slow_run])

---

CLI

tvastar run   my_agent.py:agent "Write hello.py and run it"
tvastar chat  my_agent.py:agent
tvastar serve my_agent.py:agent
tvastar info  my_agent.py:agent
tvastar logs  run_abc123

---

Deploy anywhere

One agent definition. Any target.

# AWS Lambda
from tvastar.deploy import lambda_handler
handler = lambda_handler(agent)

# GitHub Action
from tvastar.deploy import github_action
github_action(agent, on="workflow_dispatch")

# ASGI (Uvicorn, Gunicorn)
from tvastar.serving import create_app
app = create_app(agent)

---

Custom model adapter

from tvastar.model import Model
from tvastar.types import Message, ModelResponse, StopReason, TextBlock

class MyModel(Model):
    name = "my-provider"

    async def generate(self, messages, *, system=None, tools=None,
                       max_tokens=4096, temperature=1.0,
                       stop_sequences=None, thinking_level=None) -> ModelResponse:
        text = await my_api_call(messages)
        return ModelResponse(
            message=Message("assistant", [TextBlock(text=text)]),
            stop_reason=StopReason.END_TURN,
        )

---

Evals — measure agent quality

Know when your agent gets better or worse. Define test cases, run them, get a score.

import asyncio
from tvastar import EvalSuite, Case
from tvastar.eval import assert_contains, assert_ok, assert_steps_under, assert_not_contains

suite = EvalSuite(agent, concurrency=8)

suite.add(Case(
    name="writes valid Python",
    prompt="Write a function that reverses a string",
    checks=[
        assert_contains("def"),
        assert_contains("return"),
        assert_ok(),
        assert_steps_under(5),
    ],
))

suite.add(Case(
    name="does not hallucinate imports",
    prompt="Write hello world in Python",
    checks=[
        assert_contains("print"),
        assert_not_contains("import nonexistent"),
    ],
))

report = asyncio.run(suite.run())
report.print()
# ============================================================
# Eval Report  —  2/2 passed  (100%)
# Duration: 3.2s
# ============================================================
#   ✓  writes valid Python  (2.1s)
#   ✓  does not hallucinate imports  (1.1s)
# ============================================================

print(report.score)   # 1.0
print(report.passed)  # 2

Run on every PR to catch regressions before they ship.

---

Human-in-the-loop — require approval before dangerous actions

Pause an agent run and wait for a human to approve before taking an irreversible action.

from tvastar import tool
from tvastar.approval import require_approval

@tool
async def deploy_to_production(environment: str, ctx) -> str:
    """Deploy the current build to an environment."""
    await require_approval(
        f"Deploy to {environment!r}? This will affect live users.",
        timeout=120,   # seconds to wait for a human response
    )
    return do_deploy(environment)

Three backends — pick the one that fits your stack:

from tvastar.approval import ApprovalGate, set_default_gate

# CLI — prints to terminal, reads stdin (default, good for development)
set_default_gate(ApprovalGate(backend="cli"))

# Webhook — POST to your app, resolve via HTTP callback
gate = ApprovalGate(backend="webhook", webhook_url="https://myapp.com/approvals")
set_default_gate(gate)

# Event — you control resolution from outside the agent loop
pending_requests = []
gate = ApprovalGate(
    backend="event",
    on_request=lambda req: pending_requests.append(req),
)
# Later: pending_requests[0].approve() or .deny()

---

Cost tracking — know what every run costs

from tvastar.cost import cost_for_model, BudgetPolicy, BudgetExceeded

# Check cost for a model + token counts
cost = cost_for_model("claude-opus-4-6", input_tokens=1000, output_tokens=500)
print(f"${cost.usd:.4f}")   # $0.0525

# Enforce a budget — raises BudgetExceeded if the run exceeds it
agent = create_agent(
    "assistant",
    model=AnthropicModel("claude-opus-4-6"),
    budget=BudgetPolicy(max_usd=0.50, on_exceed="stop"),
)

# Or check manually after a run
result = await harness.run("Analyse this codebase")
if hasattr(result, "cost"):
    print(f"Run cost: ${result.cost.usd:.4f}")

Supported models with automatic pricing: Claude (all tiers), GPT-4o, GPT-4o-mini, o1, o3-mini, Llama via Groq, and more. Add custom rates to COST_TABLE.

---

Further reading

• API Reference — every public symbol, fully typed
• Patterns Cookbook — 20 copy-paste recipes
• CLAUDE.md — codebase map for AI assistants

---

License

MIT