tvastar 0.3.0

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

Tvastar

A Python agent harness framework. Agent = Model + Harness.

pip install tvastar

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

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

No Docker. No containers. Zero core dependencies. Real code execution 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 agent 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

---

Models

Anthropic (Claude)

from tvastar.model import AnthropicModel

model = AnthropicModel("claude-opus-4-6")   # ANTHROPIC_API_KEY env var
model = AnthropicModel("claude-sonnet-4-6", api_key="sk-ant-...")

OpenAI

from tvastar.model import OpenAIModel

model = OpenAIModel("gpt-4o")               # OPENAI_API_KEY env var

Any OpenAI-compatible provider (Groq, Ollama, Cloudflare, Together…)

model = OpenAIModel(
    model="llama-3.1-8b-instant",
    base_url="https://api.groq.com/openai/v1",
    api_key="gsk_...",
)

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

Extended thinking (reasoning models)

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

Mock (tests / offline dev)

from tvastar.model import MockModel
from tvastar.types import ToolUseBlock

model = MockModel(["Hello!", ToolUseBlock(name="add", input={"a":1,"b":2}), "Done."])

Custom provider

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,
        )

---

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 don't 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 (stateful)
sess = harness.session()
async with sess:
    await sess.prompt("Read report.txt")
    await sess.prompt("Now 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 and return a report.", result=Report)
report: Report = result.data          # validated Pydantic instance
print(report.severity)

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

---

Delegating to specialist sub-agents

Define named specialist profiles, then delegate tasks to them:

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",          # runs in isolated child session
        cancel_after=60.0,         # timeout in seconds
        result=ReviewReport,       # structured output
    )

Task delegation is capped at 4 levels deep (MAX_TASK_DEPTH) 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",       # use a specialist profile
        "cancel_after": 30.0,
        "result": SummarySchema,
    },
], concurrency=4)                    # optional semaphore cap

for r in results:
    print(r.text)

---

Workflows — durable, inspectable operations

Wrap multi-step agent work with a run ID, event log, and persistent history:

from tvastar import workflow
from tvastar.workflow import WorkflowContext

@workflow
async def summarise_document(ctx: WorkflowContext) -> dict:
    ctx.log.info("Starting summarisation", doc=ctx.payload["path"])
    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 it
run = await summarise_document.run({"path": "report.pdf"})
print(run.run_id)       # 'run_a3f9b2...'
print(run.status)       # RunStatus.COMPLETED
print(run.output)       # {'summary': '...', 'steps': 3}

# Inspect history
for past_run in summarise_document.list_runs():
    print(past_run.run_id, past_run.status, past_run.started_at)

Persist across restarts with a file-backed registry:

from tvastar.workflow import RunRegistry
registry = RunRegistry.file_backed(".tvastar-runs")

@workflow(registry=registry)
async def my_flow(ctx): ...

---

Event-driven / async dispatch

For chat bots, webhooks, and queue processors — respond immediately, run the agent in the background:

from tvastar import dispatch, dispatch_and_wait, observe_dispatch, DispatchInput

# Fire and forget — returns a dispatch_id, agent runs in background
dispatch_id = await dispatch(
    agent,
    id="user_123",                          # identifies the conversation thread
    input=DispatchInput(text=message_text, type="chat.message"),
    on_complete=lambda r: send_reply(r.text),
    on_error=lambda e: send_error(str(e)),
    cancel_after=30.0,
)

# Fire and await (when you need the result in the same context)
result = await dispatch_and_wait(agent, id="job_456", text="Process this report.")

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

Agents with the same id share a Harness — conversation history accumulates naturally across multiple dispatches.

---

Context compaction

Prevent context window exhaustion in long-running sessions:

from tvastar import CompactionPolicy

agent = create_agent(
    "long-runner",
    model=model,
    compaction=CompactionPolicy(
        max_messages=40,    # compact when history exceeds 40 messages
        keep_last=10,       # always keep the 10 most recent messages
        min_messages=20,    # don't compact below this floor
    ),
)
# Compaction fires automatically after tool turns — the model never notices.

Manual compaction:

from tvastar import compact_session
await compact_session(session, force=True)

---

Skills

Skills are reusable agent expertise defined in Markdown:

<!-- skills/code-reviewer.md -->
---
name: code-reviewer
description: Review a diff for bugs and style
tools: [read_file, grep]
---

You are a meticulous code reviewer. Inspect changed files carefully.
Report only concrete, actionable issues with file+line references.

from tvastar import SkillLibrary

agent = create_agent("dev", model=model, skills=SkillLibrary.from_dirs("skills/"))

async with sess:
    result = await sess.skill("code-reviewer", "Review changes in src/auth/")

---

Application-level file access

Stage files before the agent runs, collect outputs after — without going through the model's tool layer:

async with Harness(agent) as h:
    # Write inputs
    await h.fs.write_file("report.pdf", pdf_bytes)
    await h.fs.write_file("instructions.txt", "Summarise the PDF.")

    # Run agent
    result = await h.run("Follow instructions.txt")

    # Read outputs
    summary = await h.fs.read_file("summary.md")
    files = await h.fs.list_dir()

---

Sandboxes

from tvastar import VirtualSandbox, LocalSandbox, SecurityPolicy

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

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

---

MCP — use any published tool server

from tvastar import connect_mcp_server, default_toolset

# Spawn a local server
client = await connect_mcp_server(command="python", args=["my_mcp_server.py"])

# Or connect to a remote one
client = await connect_mcp_server(
    url="https://api.example.com/mcp",
    headers={"Authorization": "Bearer sk-..."},
)

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"))
# Checkpoints transcript + filesystem after every tool turn

# On restart — pick up where you left off
sess = harness.resume("sess_abc123") or harness.session()

---

Serving over HTTP

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

Endpoints:

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 (browser-friendly)

SSE example — stream directly in the browser or with curl:

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

---

Observability and tracing

from tvastar import Tracer, ConsoleExporter, JSONLExporter

harness = Harness(agent, tracer=Tracer([
    ConsoleExporter(),                  # human-readable to stderr
    JSONLExporter("trace.jsonl"),       # machine-readable log
]))

OpenTelemetry (Braintrust, Honeycomb, Datadog, Sentry, etc.):

pip install "tvastar[otel]"

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

---

Silent-failure detection

Agents can silently do the wrong thing — claim success over a failing run, loop forever, call a tool with bad arguments. Tvastar detects these automatically:

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

if not result.ok:                       # end_turn AND no warnings/errors
    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          # interactive REPL
tvastar serve my_agent.py:agent          # HTTP + WebSocket server
tvastar info  my_agent.py:agent          # print config
tvastar logs  run_abc123                 # inspect a workflow run

---

tvastar-fix — auto-fix failing tests

A real product built on Tvastar. An agent reads your test failures, edits the code, and iterates — then Tvastar re-runs the suite itself and only reports success on the real exit code. The agent can't lie.

pip install tvastar
export GROQ_API_KEY=...       # free tier works; or use local Ollama

tvastar-fix                   # auto-detects test framework, fixes, verifies
tvastar-fix --check           # CI mode — exit 1 if still failing

GitHub Action:

- uses: vanamayaswanth/tvastar/action@v0.2.0
  with:
    test-command: "pytest -q"
    groq-api-key: ${{ secrets.GROQ_API_KEY }}

---

Deploy anywhere

from tvastar.deploy import asgi_app, lambda_handler, serverless_handler
from my_agent import agent

app     = asgi_app(agent)           # FastAPI / Starlette — Fly, Render, Cloud Run
handler = lambda_handler(agent)     # AWS Lambda + API Gateway
fn      = serverless_handler(agent) # GCP / Azure / Vercel

---

Project layout

tvastar/
  types.py          Core dataclasses — Message, ToolUse, ModelResponse, ...
  agent.py          AgentSpec + create_agent()
  harness.py        Harness + HarnessFS + fan_out()
  session.py        Session + RunResult + the agent loop
  model/            Model ABC + Anthropic / OpenAI / Mock adapters
  tools/            @tool, ToolRegistry, ToolRetryPolicy, default_toolset()
  skills/           Markdown skill loader
  sandbox/          VirtualSandbox / LocalSandbox / external adapters
  memory/           InMemoryStore / FileStore / Memory (scoped KV)
  profiles.py       AgentProfile, define_agent_profile(), MAX_TASK_DEPTH
  workflow.py       @workflow, WorkflowContext, WorkflowRun, RunRegistry
  dispatch.py       dispatch(), dispatch_and_wait(), observe_dispatch()
  compaction.py     CompactionPolicy, compact_session()
  durable.py        Checkpointer (checkpoint / resume)
  observability.py  Tracer, Span, exporters
  detect/           Silent-failure detectors
  mcp/              MCP client (stdio + HTTP transports)
  serving/          HTTP/WebSocket/SSE server + CLI
  deploy/           ASGI / Lambda / GitHub Actions adapters
  fix/              tvastar-fix application

---

Full API reference

See docs/API.md for every public function, class, and field with full type signatures.

See docs/PATTERNS.md for copy-paste recipes.

See CLAUDE.md for the AI-optimised codebase map (module contracts, data flow, dependency graph).

---

Testing

pip install "tvastar[dev]"
pytest

---

License

MIT — vanamayaswanth/tvastar