tvastar 0.11.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, tool masking, prompt-injection 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
Inspect runs visually	Built-in trace viewer UI (tvastar ui)
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
Tool called in wrong execution phase	GovernancePolicy — tamper-proof phase enforcement
Filesystem changes need atomic rollback	harness.transaction() + sandbox snapshot/restore
Agent needs memory across sessions	tvastar.contrib.ltm — post-session LTM consolidation
Session messages balloon past 50 MB	memory_cap_mb — hard cap with auto-compaction

---

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
GovernancePolicy	Phase-based tool enforcement — declare which tools are legal per workflow phase
Finding	A structured signal from a silent-failure detector (severity + message + evidence)

---

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.

Add internet access with web_toolset() — no API key, no extra dependencies:

from tvastar import default_toolset, web_toolset

agent = create_agent(
    "researcher",
    model=model,
    tools=[*default_toolset(), *web_toolset()],
)
# Agent can now browse any URL and search the web

# Or use individually
from tvastar import web_browse, web_search

@tool
async def my_tool(url: str) -> str:
    return await web_browse.fn(url)

web_browse(url) fetches any page as clean markdown via Jina AI Reader. web_search(query) returns top search results via Jina AI Search. Both handle HTTP errors gracefully and accept a max_chars limit to protect context.

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)

---

DAG task execution — maximum parallelism

TaskGraph models work as a directed acyclic graph. Independent tasks run concurrently; a task starts the moment every dependency completes. Wall-clock time equals the critical path, not the sum of all tasks.

from tvastar import TaskGraph

graph = TaskGraph(harness)

# These three have no deps — start immediately in parallel
graph.task("leads",   "Fetch the lead list from CRM")
graph.task("pricing", "Scrape competitor pricing pages")
graph.task("news",    "Find recent news about the prospect")

# Waits for all three; their results are auto-injected into its prompt
graph.task("analyse", "Score and prioritise leads",
           depends_on=["leads", "pricing", "news"])

# These two depend on analyse but not each other — run in parallel
graph.task("emails",  "Write personalised cold emails",
           depends_on=["analyse"])
graph.task("report",  "Write executive summary",
           depends_on=["analyse"])

results = await graph.run()
print(results["emails"].text)
print(results.ok)          # True when every task finished cleanly
print(results.text)        # dict of all task outputs

Fluent chaining:

results = await (
    TaskGraph(harness)
    .task("fetch", "Fetch data")
    .task("analyse", "Analyse it", depends_on=["fetch"])
    .task("report",  "Write report", depends_on=["analyse"])
    .run()
)

Structured output per task:

graph.task("score", "Score each lead", result=LeadScores, depends_on=["fetch"])
results["score"].data  # LeadScores instance

---

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()]))

The model.generate span follows the OpenTelemetry GenAI semantic conventions — gen_ai.system, gen_ai.request.model, gen_ai.usage.input_tokens, gen_ai.response.finish_reasons, … — so traces drop into Braintrust / Honeycomb / Datadog dashboards without custom attribute mapping.

---

Trace viewer UI — inspect every run locally

Write a trace file with JSONLExporter, then open the viewer:

pip install "tvastar[serve]"
tvastar ui                          # reads tvastar-trace.jsonl in cwd
tvastar ui --trace my-run.jsonl     # custom path
tvastar ui --port 7878 --no-open    # headless / CI

Or programmatically after a run:

from tvastar import Tracer, JSONLExporter, Harness, run_ui

harness = Harness(agent, tracer=Tracer([JSONLExporter("trace.jsonl")]))
result  = await harness.run("Write and test auth.py")

# inspect in browser
run_ui("trace.jsonl", port=7878)

The viewer is a self-contained FastAPI + vanilla-JS SPA (no build step, no Node):

• Left panel — runs listed newest-first, with a green/yellow/red status dot, step count, tool-call count, and total duration
• Right panel — per-run token counts, detected findings (warnings / errors), and an expandable timeline: every model.generate, tool.invoke, and lifecycle event in order with inputs, result previews, and stop reasons
• Auto-refreshes every 5 s — watch a long run fill in live

Try it with the bundled demo (no agent run required):

python run_ui_demo.py   # generates a sample trace and opens the viewer

---

Tool masking — show the model only the tools it needs now

Exposing every tool on every turn burns context and tempts the model to reach for the wrong one. A tool_policy filters the visible toolset per turn (it can only hide available tools, never grant new ones, and never breaks a run):

from tvastar import create_agent, allow_only, deny, phases

# only one tool, ever
create_agent(..., tool_policy=allow_only("read_file"))

# everything except the dangerous one
create_agent(..., tool_policy=deny("bash"))

# research first, unlock writes once we're a few steps in
create_agent(..., tool_policy=phases({1: ["grep", "read_file"],
                                      4: ["grep", "read_file", "write_file"]}))

# or any callable: (MaskContext) -> list[str]
create_agent(..., tool_policy=lambda ctx: ["bash"] if ctx.step > 2 else [])

---

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

---

Untrusted content & prompt-injection detection

No one has solved prompt injection — so Tvastar doesn't claim to. It gives you the two honest things that genuinely help:

1. Fence untrusted content so the model treats it as data, not orders. This reduces — does not eliminate — the model following injected instructions.
2. Detect content that looks like an injection attempt and surface it as a WARNING finding (the built-in prompt_injection detector). Detection, not prevention.

from tvastar import wrap_untrusted, scan_for_injection

@tool
async def fetch(url: str) -> str:
    "Fetch a web page."
    page = await http_get(url)
    return wrap_untrusted(page, source=url)   # the model sees it as DATA

# the prompt_injection detector flags suspicious tool output automatically:
result = await harness.run("Summarise that page.")
for f in result.warnings:
    if f.detector == "prompt_injection":
        print("⚠ possible injection in tool output:", f.message)

---

Dynamic Capability Governance — lock dangerous tools to specific phases

GovernancePolicy enforces least privilege at invocation time — after the model has already decided to call a tool. Unlike masking (which is advisory), governance runs in Python code and cannot be bypassed by prompt injection.

from tvastar import create_agent, GovernancePolicy
from tvastar.approval import ApprovalGate

gov = GovernancePolicy(
    phases={
        "read":  {"grep", "read_file", "glob"},
        "write": {"grep", "read_file", "glob", "write_file", "bash"},
    },
    current_phase="read",
    # Optional — route blocked calls to a human instead of hard-blocking:
    approval_gate=ApprovalGate(backend="cli"),
)
agent = create_agent("assistant", model=..., governance=gov)

# Elevate at runtime (per-session — concurrent sessions are isolated):
gov.set_phase("write")

# Wire masking and governance together from one object:
create_agent(..., governance=gov, tool_policy=gov.as_tool_policy())

---

Transactional Sandbox — atomic rollback on failure

Wrap any session step in a harness.transaction() to guarantee that filesystem changes are rolled back if the step raises an exception.

async with harness.transaction(session) as sess:
    await sess.prompt("Refactor the auth module and run tests")
    # → if tests fail or an exception fires, the workspace rolls back atomically

Works with VirtualSandbox (< 150 ms on 1 MB) and LocalSandbox (< 500 ms on 500 KB). Both expose snapshot() / restore() for manual control too:

snap = sandbox.snapshot()
# ... do risky things ...
sandbox.restore(snap)   # reset to exactly the pre-snapshot state

---

Long-Term Memory — remember facts across sessions

tvastar.contrib.ltm consolidates conversation knowledge into a persistent LTMStore after each session and injects recalled context into the system prompt on subsequent runs. No extra dependencies needed (BM25 retrieval by default; sentence-transformers optional for semantic search).

from tvastar.contrib.ltm import LTMStore
from tvastar.memory.store import FileStore

ltm = LTMStore(FileStore(".ltm"))

# Wire retrieval into the system prompt — recalled per turn, keyed on user intent
agent = create_agent("assistant", model=..., system_prompt_hook=ltm.as_hook())

# After the session completes, persist what the agent learned
result = await Harness(agent).run("Fix the flaky auth test")
await ltm.consolidate(result, model, session_id="fix-auth-001")

# Next session — the agent automatically recalls relevant past knowledge
result2 = await Harness(agent).run("The auth test is flaky again")

---

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
tvastar ui    --trace tvastar-trace.jsonl   # local trace viewer
tvastar bench my_agent.py:agent --suite swe-lite --max-tasks 10

---

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.

---

Benchmarks — measure quality against the real world

EvalSuite measures against your checks. BenchSuite measures against standardised, external task sets — the difference between testing whether your code works and testing whether your agent works on real software engineering problems.

import asyncio
from tvastar import create_agent, BenchSuite, swe_bench_tasks, default_toolset
from tvastar.model import AnthropicModel

agent = create_agent("coder", model=AnthropicModel(), tools=default_toolset())
suite = BenchSuite(agent, concurrency=4)
suite.add_many(swe_bench_tasks(split="lite", max_tasks=10))   # needs: pip install datasets
report = asyncio.run(suite.run())
report.print()
# ═══════════════════════════════════════════════════════════════
# Benchmark Report
#   Resolved : 7/10  (70.0%)
#   Duration : 142.3s
# ═══════════════════════════════════════════════════════════════

Or from the CLI:

tvastar bench agent.py:agent --suite swe-lite --max-tasks 50 --out report.json

Local JSONL — bring your own benchmark in SWE-bench format:

tvastar bench agent.py:agent --suite ./my_tasks.jsonl --max-tasks 20

Verification runs real pytest on the workspace — not the model's say-so. Results are labelled swe_lite_local to distinguish them from the official Docker-based harness numbers. Use the official harness for published comparisons; use this for rapid iteration.

---

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.

---

What we're building

Tvastar is the engine. Every product below is built on top of it — same harness, same tools, same deploy model. Framework features get added only when a product needs them.

---

✅ tvastar-fix — Auto-repair failing tests

Shipped. The reference implementation.

Your CI fails. tvastar-fix runs the agent, edits the source, re-runs the suite itself, and pushes the fix — without you touching a line. Verification is a real exit code, never the model's claim.

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

---

✅ tvastar-outbound — AI outbound sales agent

Shipped v0.9.0.

Give it a CSV of leads. It researches each one in parallel (company site, news, LinkedIn via web_browse + web_search), scores and prioritises them with TaskGraph, writes a personalised cold email for each, waits for your approval via ApprovalGate, then sends. Full audit trail in the trace viewer.

pip install tvastar
tvastar-outbound --csv leads.csv --icp "B2B SaaS, 50+ employees" \
    --sender-name "Jane" --sender-company "Acme" --sender-email [e-mail] \
    --min-score 0.6 --dry-run

Or programmatically:

from tvastar.outbound import run_campaign
from tvastar.model import AnthropicModel

result = await run_campaign(
    "leads.csv",
    model=AnthropicModel("claude-sonnet-4-5"),
    icp="B2B SaaS companies with 50+ employees struggling with developer productivity",
    sender_name="Jane Smith",
    sender_company="Acme",
    sender_email="[e-mail]",
    min_score=0.6,
)
print(f"Sent {result.sent}/{result.leads_qualified} emails.")

Why Tvastar is the right engine:

• TaskGraph researches all leads in parallel — 50 leads in wall-clock time of 1
• web_browse + web_search — no external scraping service needed
• ApprovalGate — human reviews every draft before anything goes out
• BudgetPolicy — hard cost ceiling per campaign
• JSONLExporter + tvastar ui — see every email and every research step

---

🔒 tvastar-comply — PII / PFI / PHI compliance layer

Enterprise-grade data protection baked into the agent loop.

Healthcare, finance, and legal companies cannot use AI agents on real customer data without a compliance layer. tvastar-comply solves this at the harness level — not as a bolt-on service, but as a first-class part of every agent run.

What it handles:

Type	Examples	Regulation
PII	Name, email, phone, address, SSN	GDPR, CCPA
PFI	Credit cards, bank accounts, tax records	PCI-DSS, GLBA
PHI	Medical records, diagnoses, prescriptions	HIPAA

How it works — redact before LLM, rehydrate after:

from tvastar.comply import ComplyPolicy

agent = create_agent(
    "support",
    model=AnthropicModel(),
    comply=ComplyPolicy(
        scan=["pii", "phi", "pfi"],  # what to detect
        action="redact",              # redact | block | audit
        vault="local",               # token vault stays on your machine
    ),
)

# Input:  "John Smith, DOB 1990-01-01, diagnosis: diabetes"
# → LLM sees: "[NAME_1], DOB [DATE_1], diagnosis: [CONDITION_1]"
# → LLM responds: "I've updated [NAME_1]'s care plan"
# → Output: "I've updated John Smith's care plan"

Compliance audit trail built in:

result = await harness.run("Process patient intake form")
result.comply_report  # what was found, redacted, timestamp, hash — per run

Why this matters for every other Tvastar product:

• tvastar-outbound processes lead PII → GDPR requires it
• tvastar-support handles customer data → CCPA requires it
• tvastar-devops reads logs that may contain secrets → security requires it

One comply=ComplyPolicy(...) line makes any agent enterprise-ready. The token vault is local — no PII ever leaves your infrastructure.

---

📋 tvastar-review — GitHub PR review bot

Coming after tvastar-outbound.

Webhook fires on PR open → agent reads the diff → posts inline comments → flags shallow or unverified completions using the built-in detectors. Ships as a zero-config GitHub Action.

- uses: vanamayaswanth/tvastar-review@v1
  with:
    model: claude-sonnet-4-6

---

🛠 tvastar-devops — Production auto-heal agent

Extending tvastar-fix to live systems.

Log watcher detects anomaly → agent diagnoses root cause → runs bash fix → verifies with a real exit code → pages you only if it cannot fix it. Same "verify with real signals" principle as tvastar-fix, extended to production incidents.

---

💬 tvastar-support — Customer support agent

Multi-platform, persistent, production-ready.

One session per user, memory across conversations, simultaneous Telegram / Slack / email. dispatch() per inbound message, on_complete sends the reply. Human escalation via ApprovalGate when confidence is low.

---

🔍 tvastar-research — Competitive intel agent

Parallel web research → structured report.

Describe what you want to know. Agent fans out across sources with fan_out(), synthesises with structured output (result=), delivers a report. VCs, analysts, marketing teams.

---

Roadmap

Products ship first. Framework features get added only when a product needs them.

Milestone	What ships	Status
Web tools	web_browse + web_search — Jina AI, zero deps	✅ v0.8.1
DAG execution	TaskGraph — parallel tasks, critical path only	✅ v0.8.0
tvastar-outbound	Outbound sales agent — research → score → email → send	✅ v0.9.0
SOTA safety	Governance, transactions, LTM, memory cap, OpenAI retry	✅ v0.10.0
tvastar-comply	PII / PFI / PHI redaction layer — GDPR, HIPAA, PCI-DSS	🔒 v0.10.1
Platform gateway	Telegram + cron — added when outbound needs notifications	📋 v1.0.0
Skill learning loop	Agent writes Skills from successful runs; FTS memory	📋 v1.1.0
tvastar-review	GitHub PR bot — diff → inline comments → GitHub Action	📋 v1.2.0
tvastar-devops	Auto-heal production incidents	📋 v1.3.0
tvastar-support	Multi-platform customer support agent	📋 v1.4.0
Hosted platform	Cloud-hosted harness, product dashboard, skill marketplace	📋 v2.0.0

Framework features are only added when a product needs them — not to match a checklist. tvastar-comply unlocks healthcare, finance, and legal — the highest-value enterprise markets.

---

Further reading

• API Reference — every public symbol, fully typed
• Patterns Cookbook — 20 copy-paste recipes
• 12-Factor Agents map — how Tvastar maps to the production checklist (honest verdicts)
• AGENTS.md — contributor guide for working in this repo
• CLAUDE.md — codebase map for AI assistants

---

License

MIT