Metadata-Version: 2.4
Name: tvastar
Version: 0.3.0
Summary: Tvastar — a programmable agent harness framework for Python. Agent = Model + Harness.
Project-URL: Homepage, https://github.com/vanamayaswanth/tvastar
Project-URL: Repository, https://github.com/vanamayaswanth/tvastar
Project-URL: Issues, https://github.com/vanamayaswanth/tvastar/issues
Project-URL: Changelog, https://github.com/vanamayaswanth/tvastar/blob/main/CHANGELOG.md
Author-email: vanamayaswanth <vanamayaswanth@gmail.com>
License: MIT
License-File: LICENSE
Keywords: agents,ai,claude,harness,llm,mcp,model-context-protocol,sandbox,skills,tools
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Requires-Python: >=3.10
Provides-Extra: all
Requires-Dist: anthropic>=0.40.0; extra == 'all'
Requires-Dist: fastapi>=0.110.0; extra == 'all'
Requires-Dist: openai>=1.40.0; extra == 'all'
Requires-Dist: opentelemetry-api>=1.20.0; extra == 'all'
Requires-Dist: opentelemetry-sdk>=1.20.0; extra == 'all'
Requires-Dist: uvicorn[standard]>=0.27.0; extra == 'all'
Requires-Dist: websockets>=12.0; extra == 'all'
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40.0; extra == 'anthropic'
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: openai
Requires-Dist: openai>=1.40.0; extra == 'openai'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20.0; extra == 'otel'
Requires-Dist: opentelemetry-sdk>=1.20.0; extra == 'otel'
Provides-Extra: serve
Requires-Dist: fastapi>=0.110.0; extra == 'serve'
Requires-Dist: uvicorn[standard]>=0.27.0; extra == 'serve'
Requires-Dist: websockets>=12.0; extra == 'serve'
Description-Content-Type: text/markdown

# Tvastar

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

```bash
pip install tvastar
```

```python
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

```bash
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)

```python
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

```python
from tvastar.model import OpenAIModel

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

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

```python
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)

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

### Mock (tests / offline dev)

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

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

### Custom provider

```python
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

```python
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:

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

---

## Sessions

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

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

---

## Skills

Skills are reusable agent expertise defined in Markdown:

```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.
```

```python
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:

```python
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

```python
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

```python
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

```python
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

```bash
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:

```bash
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

```python
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.):

```bash
pip install "tvastar[otel]"
```

```python
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:

```python
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:

```python
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

```bash
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.

```bash
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:**

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

---

## Deploy anywhere

```python
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](docs/API.md) for every public function, class, and field with full type signatures.

See [docs/PATTERNS.md](docs/PATTERNS.md) for copy-paste recipes.

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

---

## Testing

```bash
pip install "tvastar[dev]"
pytest
```

---

## License

MIT — [vanamayaswanth/tvastar](https://github.com/vanamayaswanth/tvastar)
