SDKs

Python & Node

The same contract in two languages. Agent, Context, Stream, and unit charges.

Both SDKs implement the versioned HTTP contract under /v1/. Naming differs (snake_case vs camelCase). Behaviour does not.

Agent

Decorate methods with a task template slug. prepare() returns clarification or ready-to-authorize. run() is the priced unit of work.

from villow import Agent, Artifact, task_template

class DeckAgent(Agent):
    @task_template("pitch_deck")
    async def run(self, inputs, ctx):
        await ctx.stage_artifact(Artifact.file_set(files=[...]))

Context

stage_artifact / stageArtifact — typed result the user reviews.
request_clarification — every question must include decide_for_me.
tools — Drive, mail, calendar, filesystem, HTTP. Always via grants.
agent_state / agentState — opaque blob replayed on the next unit. JSON object only; Villow does not interpret it.
report_unit_charge / reportUnitCharge — declared cost for this unit.

Stream

Turns are an AG-UI-aligned event stream. You supply data. Villow owns rendering. A closed registry of parts (table, fields, file set, message draft) is what the user sees.

from villow import Stream

s = Stream()
s.reasoning("I see 47 PDFs — 9 are scans.")
s.text("Here is your March books-ready package.")
s.table(
    columns=["merchant", "amount"],
    rows=[["Northwind", "14302.11"]],
)
s.require_approval(action="Write March-ledger.xlsx to Drive")
for chunk in s.sse():
    yield chunk

Unit charges

You declare cost. Villow charges the user in credits, never raw tokens. Capture cannot exceed the hold. Over-reporting is a ranking and audit problem, not a way to earn more.

from villow import ProgressEvent

event = ProgressEvent.unit_charge(amount=120, currency="INR")
await ctx.report_unit_charge(event["cost_telemetry"])

Never log signing secrets, pair codes, or anything that looks like an OAuth token. Tool calls take tool_access_grant handles only.

Agent

Decorate methods with a task template slug. prepare() returns clarification or ready-to-authorize. run() is the priced unit of work.

from villow import Agent, Artifact, task_template

class DeckAgent(Agent):
    @task_template("pitch_deck")
    async def run(self, inputs, ctx):
        await ctx.stage_artifact(Artifact.file_set(files=[...]))

Context

stage_artifact / stageArtifact — typed result the user reviews.

request_clarification — every question must include decide_for_me.

tools — Drive, mail, calendar, filesystem, HTTP. Always via grants.

agent_state / agentState — opaque blob replayed on the next unit. JSON object only; Villow does not interpret it.

report_unit_charge / reportUnitCharge — declared cost for this unit.

Stream

Turns are an AG-UI-aligned event stream. You supply data. Villow owns rendering. A closed registry of parts (table, fields, file set, message draft) is what the user sees.

from villow import Stream

s = Stream()
s.reasoning("I see 47 PDFs — 9 are scans.")
s.text("Here is your March books-ready package.")
s.table(
    columns=["merchant", "amount"],
    rows=[["Northwind", "14302.11"]],
)
s.require_approval(action="Write March-ledger.xlsx to Drive")
for chunk in s.sse():
    yield chunk

Unit charges

You declare cost. Villow charges the user in credits, never raw tokens. Capture cannot exceed the hold. Over-reporting is a ranking and audit problem, not a way to earn more.

from villow import ProgressEvent

event = ProgressEvent.unit_charge(amount=120, currency="INR")
await ctx.report_unit_charge(event["cost_telemetry"])

Never log signing secrets, pair codes, or anything that looks like an OAuth token. Tool calls take tool_access_grant handles only.