Taskwarden

A durable, owner-approved AI team for a business.

Taskwarden is a Python harness for running a small team of AI agents that do real work for a business (texting customers, booking, refunding, posting) on a server, safely. The owner decides what each agent may do alone and what must ask first. Anything that asks waits on a queue in your database until the owner approves, edits or declines it, then carries on where it stopped, even after a restart.

It runs the AI team of Perbene in production, for many businesses at once.

pip install taskwarden

Why another agent framework

Most frameworks are built for one person at a keyboard, or for a demo. A business's team runs unattended, for many businesses at once, and its mistakes reach customers. Taskwarden is built around the questions that come up then:

How it fits

owner request ──▶ loop.run_task ──▶ model ──▶ tool call ──▶ gate.decide
                                                              │
                         allow ◀───────────────┬──────────────┤
                           │                   │              ▼
                       run the tool        observe        ask: park the task
                           │               (record)        (pending, in your DB)
                           ▼                                  │
                    next round / report        owner: approve / edit / decline
                                               decisions.claim + decisions.apply
                                                              │
                                                 queued ──▶ sweeper resumes it

Quick start

examples/shop.py is a complete app: a web shop whose agent reads orders on its own and refunds only with the owner's click. It runs offline (SQLite, a scripted model):

pip install "taskwarden[dev]"
python examples/shop.py

owner: Refund the customer who paid $19.50 today.
run:   waiting
waits: refund {'order_id': 'A7'}
owner: approved, with order A8 instead of A7
run:   done
...
chat: [lead] Done: I refunded order A8, $19.50.

The shape of a host:

from taskwarden import decisions, loop, sweeper, tools
from taskwarden.host import BaseHost
from taskwarden.models import MessageColumns, TaskColumns

class Task(Base, TaskColumns):
    __tablename__ = "team_tasks"

class Message(Base, MessageColumns):
    __tablename__ = "team_messages"

class ShopHost(BaseHost):
    Task, Message = Task, Message
    chief = "lead"
    agents = {"lead": Agent("lead", "Sam", "Shop assistant", ("read_orders", "refund"))}
    tools = {
        "read_orders": tools.Tool("read_orders", tools.READ, schema, read_orders),
        "refund": tools.Tool("refund", tools.SPEND, schema, refund),
    }

    async def llm(self, db, organization_id):
        return openai.AsyncOpenAI(), "your-model"   # any OpenAI-compatible client

    async def system_prompt(self, db, task, agent, name, *rest) -> str:
        return f"You are {name}, the {agent.title} of a small web shop."

await loop.run_task(host, db, task, actor="owner")          # the owner's request
await decisions.apply(host, db, org, await decisions.claim(host, db, org, task.id, "approve"),
                      "approve")                             # their click, any time later
await sweeper.sweep(host, Session, redis, now=now, run=run) # on a timer: resumes the rest

BaseHost leaves out what only your app can know. Override it as you grow: sent_key / reach_key (turn on the duplicate-send guard), notify_parked (push, SMS or email the owner), refuse_run (credits), context (the business's time zone), audit (store the trail), mark (live progress), thought (the model's reasoning, as it works), take_messages (what the owner writes while a task runs joins it), stop_asked (the owner's Stop: the run ends at its next step and says what it did), clock (how the time reads; it rides on the first message of each request, never in the system prompt, so the prompt stays cached).

Risk levels

Requirements

Python 3.11+, SQLAlchemy 2 (async), PyYAML. Postgres in production (SQLite works for trying it out), Redis for the sweeper's lock. Any OpenAI-compatible chat model with tool calls.

Tests

pip install -e ".[dev]"
pytest

License

Apache-2.0. Built by DAHO.