Skip to content

Govern an existing client

You don't need tasks to start. An app that already calls a model through the OpenAI or Anthropic SDK can be governed in three lines, without changing the code that uses the client. Every call is then traced, costed, budgeted and guarded. That is adoption level L0 (Adoption levels).

The three lines

import harnessweaver as hw                                    # 1

harness = hw.load("harness.yaml")                             # 2
client = harness.instrument(AsyncOpenAI(base_url=GATEWAY))    # 3
# The rest of the app uses `client` exactly as before.

A complete example

Example 01 runs this offline. Its spec has no tasks, just the extraction profile's guards and a budget:

instrument.yaml
# The spec for l0_instrument.py: an existing app that governs the client it
# already has, before it moves to tasks. The extraction profile's PII redaction
# and secrets check apply to every call. There are no tasks and no business
# rules here: the client's answers are text, and harness.yaml's rules are about
# a typed Invoice (a rule that can't be checked fails closed, ADR-0023).
apiVersion: harnessweaver/v1
kind: Harness
metadata:
  name: invoice-extraction-l0
  owner: finance-platform
  risk_tier: internal
extends: ["profile://extraction"]

budgets:
  per_run: { max_usd: 0.02 }   # each instrumented call is one run

And the app instruments the client it already has:

l0_instrument.py (excerpt)
async def main() -> Invoice:
    harness = hw.load(SPEC)  # 2
    client = harness.instrument(existing_client(), tenant="acme")  # 3

    # Unchanged app code from here on.
    reply = await client.chat.completions.create(
        model="fast-chat",
        messages=[
            {"role": "system", "content": "Extract the invoice as JSON."},
            {"role": "user", "content": INVOICE_TEXT},
        ],
    )
    invoice = Invoice.model_validate_json(reply.choices[0].message.content or "")
    print(f"invoice: {invoice.invoice_number}, total {invoice.total:.2f} {invoice.currency}")
    for line in RECEIVED[-1].splitlines():
        if "Billed to" in line or "Call" in line:
            print(f"the gateway saw: {line.strip()}")
    return invoice

$ HARNESSWEAVER_ENV=offline uv run python examples/01_invoice_extraction/l0_instrument.py
invoice: INV-2026-0042, total 1860.00 EUR
the gateway saw: Billed to: Acme Corp, [EMAIL]
the gateway saw: Questions? Call [PHONE]

The app's call is unchanged. The PII guard redacted the customer's details before the request left the process, so the gateway saw only placeholders, and the answer was checked for secrets before the app got it.

What instrument does

  • harness.instrument(client) finds an adapter from the client's type: it loads the plugin whose namespace is the client's top-level package (openai, anthropic) and asks it for an adapter. adapter= names one explicitly. The adapter follows the client, not models.provider, so an app using the Anthropic SDK needs harnessweaver-anthropic whichever provider its spec names.
  • It governs chat.completions.create on openai.OpenAI and AsyncOpenAI, and messages.create on anthropic.Anthropic and AsyncAnthropic. Other methods pass through.
  • Outside a run, each call becomes a small run of its own: run.start, input.before on the message texts, model.before, model.after, output.before on the answer, run.end. So the budget, the guards and the trace apply per call.
  • Inside a run or session, a call is one governed model call of that run.
  • Redactions are written back: the API receives the redacted request, and the app gets the answer as the output guards left it.
  • A request the model declines raises RunBlocked with the code model_refusal, rather than returning an answer the app must inspect.
  • If the baseline or the budget needs a hook the adapter can't reach, instrument() fails instead of running ungoverned (RULES R4).
  • Streaming through an instrumented client is planned. Until then it raises, rather than running ungoverned.

Pass tenant= and task= to say whom the calls act for, and which task's budget and guards apply:

client = harness.instrument(AsyncOpenAI(base_url=GATEWAY), tenant="acme", task="triage")

Agent frameworks

harness.wrap(agent, task=...) does the same for an agent from another framework. For Pydantic AI it governs every model request, tool call and the final output. A call that needs approval escalates, because the harness can't resume the app's code. To get approvals that suspend and resume, run the agent as a task with the pydantic_ai.agent runner instead (Runners, deciders and adapters).

Next steps

Once calls are governed, move the work that matters into tasks one at a time: typed output gets you to L1, policy to L2. Integration styles compares Wrap, Session and Run.