HarnessWeaver¶

The composable AI harness framework. Weave your own AI harness: bring your models, choose your plugins, define your execution.

Source Code: github.com/Wise-DreamerZ/HarnessWeaver

HarnessWeaver is a framework for building your own AI harness: the controls every AI application needs around its models, declared in one harness.yaml and applied at 13 fixed hook points, whatever executor your app already uses.

The key features are:

Declarative: tracing, cost, budgets, guardrails, typed output, tool governance, policy, approvals, routing, retrieval governance and evals, all in one spec. Your code stays one call.
Non-invasive: keep the executor you have, whether raw SDK calls, Pydantic AI, the Vercel AI SDK or your own pipeline. Govern an existing client in three lines.
Safe by default: the org baseline can't be removed by an app, controls fail closed, and no prompt or output reaches a log or a span unredacted.
Composable: every capability is a module behind a port, with providers you swap, and plugins anyone can write in their own repository.
Polyglot: one spec serves Python, TypeScript, and any language through the harnessd sidecar.
Testable: everything runs offline against fakes, with no network and no model spend, and every decision is in a trace your tests can check.
Standards-based: OpenTelemetry GenAI conventions, JSON Schema contracts, an OpenAPI sidecar, and a CEL subset for rules.

Requirements¶

Python 3.12 or later. The core depends only on Pydantic, the OpenTelemetry API, PyYAML, anyio and loguru. Everything else is a plugin.

Installation¶

$ uv add "harnessweaver[openai]"

Example¶

Create it¶

Create a harness.yaml:

harness.yaml

apiVersion: harnessweaver/v1
kind: Harness
metadata:
  name: support-desk
  owner: support-team

models:
  provider: openai.chat
  base_url: http://localhost:4000
  api_key_env: LITELLM_API_KEY
  default: fast
  aliases: { fast: fast-chat }

tasks:
  triage:
    runner: direct
    output: str

And a main.py:

main.py

import asyncio

import harnessweaver as hw


async def main() -> None:
    harness = hw.load("harness.yaml")
    result = await harness.run("triage", "I was charged twice for my Pro plan. Refund one?")
    print(result.status.value)
    print(result.output)


if __name__ == "__main__":
    asyncio.run(main())

To run without a model gateway, add harness.offline.yaml, which swaps the model for a scripted one:

harness.offline.yaml

models:
  provider: scripted
  replies:
    - text: "billing, priority 2: charged twice for the Pro plan, wants one refund."

Run it¶

$ HARNESSWEAVER_ENV=offline uv run python main.py
completed
billing, priority 2: charged twice for the Pro plan, wants one refund.

Check it¶

Every run keeps a trace of the hook points it passed and every decision made, with no content in it. Print it with harnessweaver.testing.render_trace(result):

   0 hook    run.start         harness -> continue
   1 hook    input.before      harness -> continue
   2 hook    model.before      harness -> continue
   3 note    model.after       harness.budget: model usage
   4 hook    model.after       harness -> continue
   5 hook    output.before     harness -> continue
   6 hook    run.end           harness -> continue

Example upgrade¶

Now give the task typed output, the extraction profile's guards, a business rule and a budget, all in the spec:

harness.yaml

apiVersion: harnessweaver/v1
kind: Harness
metadata:
  name: support-desk
  owner: support-team
extends: ["profile://extraction"]

models:
  provider: openai.chat
  base_url: http://localhost:4000
  api_key_env: LITELLM_API_KEY
  default: fast
  aliases: { fast: fast-chat }

tasks:
  triage:
    runner: direct
    prompt: triage@2
    input: helpdesk.models:Ticket
    output: helpdesk.models:Triage
    budget: { max_model_calls: 3, max_usd: 0.01, deadline: 30s }

validation:
  rules:
    - id: bugs-are-urgent
      when: "output.category == 'bug' && output.priority > 2"
      message: "a bug is priority 1 or 2"

The types are plain Pydantic models:

helpdesk/models.py

"""What goes into the support desk's tasks and what comes out."""

from typing import Literal

from pydantic import BaseModel, Field


class Ticket(BaseModel):
    """A message a customer sent to the support desk."""

    text: str


class Triage(BaseModel):
    """How the desk should handle a ticket."""

    category: Literal["billing", "bug", "account", "other"]
    priority: int = Field(ge=1, le=4, description="1 is the most urgent.")
    summary: str = Field(description="One sentence: what the customer wants.")

The app is still one call:

main.py

import asyncio

from helpdesk.models import Ticket, Triage

import harnessweaver as hw


async def main() -> None:
    harness = hw.load("harness.yaml")
    ticket = Ticket(text="I was charged twice for my Pro plan. Refund one? ann@example.com")
    result = await harness.run("triage", ticket)
    if result.ok:
        triage = Triage.model_validate(result.output)
        print(f"{triage.category}, priority {triage.priority}: {triage.summary}")
    print(result.status.value)
    for event in result.trace:
        if event.code:
            print(f"{event.source} ({event.code}): {event.reason}")


if __name__ == "__main__":
    asyncio.run(main())

$ HARNESSWEAVER_ENV=offline uv run python main.py
billing, priority 2: Charged twice for the Pro plan and wants one payment refunded.
completed
harness.guard (guard_fired): guardrails.input[extraction-pii] redact

The customer's e-mail address was redacted before the model saw it, and the guard's firing is in the trace.

Recap¶

With a few lines of YAML, and no change to how the app calls the harness, the task now has:

Typed output: the answer is parsed into Triage, and the category and priority are checked against the type.
Repair: an answer that doesn't parse, or breaks bugs-are-urgent, is sent back to the model with the reason, once, and then escalated to a person.
Guards: e-mail addresses, phone numbers and card numbers are redacted from the input, and secrets are blocked in the output.
A budget: at most three model calls, one cent and thirty seconds per run, checked before every call.
A trace: every guard, rule, repair and budget check, without any content.

And the same spec, extended by your org's baseline, can't have its required controls switched off.

The Tutorial builds a complete support desk this way, with routing, retrieval, tools, policy, approvals, state, observability, tests and evals.

Already have an app?¶

Govern the client it already uses, without changing the code that calls it:

client = harness.instrument(AsyncOpenAI(base_url="http://localhost:4000"), tenant="acme")
reply = await client.chat.completions.create(model="fast-chat", messages=messages)   # traced, costed, budgeted, guarded

See Govern an existing client.

Not an agent framework¶

HarnessWeaver isn't an agent framework, a prompt library, a vector database or a model host, and it never does the reasoning itself (RULES R12). It puts controls around the executor your app already uses. Alternatives and comparisons compares it with the tools around it.

Dependencies¶

The core, harnessweaver, depends on:

pydantic: every spec, contract and payload is a Pydantic model.
opentelemetry-api: spans and metrics, without an SDK unless you add one.
pyyaml: specs are YAML 1.2.
anyio: async everywhere.
loguru: logs, silent until your app enables them.

Optional plugins, installed with extras such as harnessweaver[openai]:

harnessweaver-openai: any OpenAI-compatible endpoint (LiteLLM Proxy, OpenAI, vLLM, Ollama).
harnessweaver-anthropic: Claude, through the Messages API.
harnessweaver-pydantic-ai: Pydantic AI agents, as a runner or wrapped.
harnessweaver-otel: OpenTelemetry SDK setup and export.
harnessweaver-cel: full CEL for policy rules.
harnessweaver-monty: a sandbox for code-executing tools.

And the tools: harnessweaver-cli (the harnessweaver command) and harnessweaver-sidecar (harnessd).

License¶

This project is licensed under the terms of the Apache-2.0 license.