Skip to content
aarness.
DocsServer & SDKs
On this page

TypeScript and Python SDKs

The SDKs wrap the server API in typed clients. Use them to run turns from your own tools, tests or services.

TypeScript Python
Package @haarness/sdk haarness-sdk
Import @haarness/sdk haarness_sdk
Needs Node.js 22+ Python 3.10+
Style Promises, async iterators Sync Haarness and async AsyncHaarness

Install

The SDKs are not on npm or PyPI yet. Install them from a checkout of the Haarness repository:

cd haarness/sdks/typescript && npm ci && npm run build
cd ~/code/my-app && npm install /path/to/haarness/sdks/typescript
python -m pip install /path/to/haarness/sdks/python

Ask one question

Start a server in your project with haarness serve, then:

import { Haarness } from "@haarness/sdk";

const client = new Haarness({ baseUrl: "http://127.0.0.1:4096" });
try {
  const thread = await client.createThread({ title: "Quick question" });
  const result = await client.chat(thread.id, "What does this project do?");
  console.log(result.turn.status);
  console.log(result.text);
} finally {
  client.close();
}
from haarness_sdk import Haarness

with Haarness("http://127.0.0.1:4096") as client:
    thread = client.create_thread({"title": "Quick question"})
    result = client.chat(thread["id"], "What does this project do?")
    print(result["turn"]["status"])
    print(result["text"])

You see completed and the answer. chat() starts a turn, waits up to five minutes for it to finish, and returns that turn's text, its items and the turn itself.

If the turn is still going after five minutes, chat() raises ApiError with status 409 and the turn keeps running. For longer work, start the turn and follow the event stream, as in the next example.

Use the server's root URL, not /v1. For a password-protected server, pass password (and username if it is not haarness). A bearer token is read from HAARNESS_SERVER_TOKEN, or pass token.

Stream the answer and handle approvals

chat() waits silently, and a turn that needs your approval waits with it. This complete example prints the answer as it streams and asks you in the terminal when the agent wants permission.

import { randomUUID } from "node:crypto";
import { createInterface } from "node:readline/promises";
import { Haarness } from "@haarness/sdk";

const prompt = process.argv.slice(2).join(" ") || "Summarise README.md";
const client = new Haarness({ baseUrl: "http://127.0.0.1:4096" });
const terminal = createInterface({
  input: process.stdin,
  output: process.stdout,
});

try {
  const thread = await client.createThread({ title: "SDK example" });
  const { eventCursor } = await client.info();
  const turn = await client.startTurn(thread.id, {
    requestId: randomUUID(),
    text: prompt,
  });

  const answered = new Set<string>();
  for await (const event of client.events({ after: eventCursor })) {
    if (!("eventId" in event) || event.rootThreadId !== thread.id) continue;
    const payload = event.payload as Record<string, any>;
    if (event.type === "turn.text_delta" && event.turnId === turn.id) {
      process.stdout.write(payload.delta);
    } else if (event.type === "permission.requested") {
      if (answered.has(payload.permissionId)) continue;
      answered.add(payload.permissionId);
      const answer = await terminal.question(
        `\nAllow ${payload.operation}? [y/N] `,
      );
      await client.replyPermission(payload.permissionId, {
        decision: answer.trim() === "y" ? "once" : "reject",
      });
    } else if (event.type === "question.requested") {
      const choices = payload.choices?.length
        ? ` (${payload.choices.join(" / ")})`
        : "";
      const answer = await terminal.question(`\n${payload.prompt}${choices} `);
      await client.replyQuestion(payload.questionId, { answer });
    } else if (event.type === "turn.finished" && payload.id === turn.id) {
      console.log(`\n[${payload.status}]`);
      break;
    }
  }
} finally {
  terminal.close();
  client.close();
}
import sys
import uuid
from contextlib import closing

from haarness_sdk import Haarness

prompt = " ".join(sys.argv[1:]) or "Summarise README.md"

with Haarness("http://127.0.0.1:4096") as client:
    thread = client.create_thread({"title": "SDK example"})
    cursor = client.info()["eventCursor"]
    turn = client.start_turn(
        thread["id"], {"requestId": str(uuid.uuid4()), "text": prompt}
    )

    answered = set()
    with closing(client.events(after=cursor)) as events:
        for event in events:
            if event.get("rootThreadId") != thread["id"]:
                continue
            kind, payload = event["type"], event["payload"]
            if kind == "turn.text_delta" and event["turnId"] == turn["id"]:
                print(payload["delta"], end="", flush=True)
            elif kind == "permission.requested":
                if payload["permissionId"] in answered:
                    continue
                answered.add(payload["permissionId"])
                answer = input(f"\nAllow {payload['operation']}? [y/N] ")
                decision = "once" if answer.strip() == "y" else "reject"
                client.reply_permission(
                    payload["permissionId"], {"decision": decision}
                )
            elif kind == "question.requested":
                choices = " / ".join(payload.get("choices") or [])
                hint = f"({choices}) " if choices else ""
                answer = input(f"\n{payload['prompt']} {hint}")
                client.reply_question(payload["questionId"], {"answer": answer})
            elif kind == "turn.finished" and payload["id"] == turn["id"]:
                print(f"\n[{payload['status']}]")
                break

Save either as a file and run it with a prompt, for example npx tsx ask.ts "List the TODO comments" or python ask.py "List the TODO comments".

How it works:

  • info() returns the current event position. Streaming from there shows only new events.
  • startTurn returns at once. Each turn needs a new requestId.
  • The stream carries every chat on the server, so the loop keeps only this one, subagents included (rootThreadId).
  • A request can be announced again, after a reconnect for example, so the loop answers each permissionId once.
  • The agent may also ask you something (question.requested), with or without choices to pick from. The turn waits for your answer.
  • Closing the stream or the client never stops the turn. Use interruptTurn(turn.id) (interrupt_turn in Python) to stop it.

Choose how much a chat asks

New chats start in Auto: a reviewer model on the server approves routine requests once, blocks risky ones with a reason the agent reads, and leaves the rest to you. Only those reach your loop. Switch a chat to Review to be asked about everything, or to Yolo to approve everything without asking (deny rules still hold):

await client.setApprovalMode(thread.id, { mode: "review" });
console.log((await client.getApprovalMode(thread.id)).mode); // "review"
client.set_approval_mode(thread["id"], {"mode": "review"})
print(client.get_approval_mode(thread["id"])["mode"])  # review

The mode belongs to the chat, so the terminal UI and Desktop see the change too. (await client.info()).defaultApprovalMode tells you which mode new chats on this server start in. See Review, Auto or Yolo for what each one allows.

The SDKs never approve anything on their own: a request nobody answers keeps its turn waiting. For a script that runs with nobody watching:

  • Send the turn with tools: { question: false } so the agent does not stop to ask you something.
  • Answer a request you do not want with { decision: "refuse", feedback: "…" }: that request alone is refused, and the agent reads it as Haarness's refusal rather than yours.

Async Python

import asyncio
from haarness_sdk import AsyncHaarness

async def main():
    async with AsyncHaarness("http://127.0.0.1:4096") as client:
        thread = await client.create_thread({"title": "Async"})
        result = await client.chat(thread["id"], "What does this project do?")
        print(result["text"])

asyncio.run(main())

Iterate events with async for, and close an early-exited stream with contextlib.aclosing. Cancelling the task cancels the request in flight. AsyncLocalServer is the async form of the LocalServer below.

Start a private server from a script

LocalServer starts a server for a folder, gives you a client, and stops the server when you are done. It uses a random port and token, so nothing else can connect.

import { LocalServer } from "@haarness/sdk";

const server = await LocalServer.start({ workspace: "/path/to/project" });
try {
  const thread = await server.client.createThread({});
  console.log((await server.client.chat(thread.id, "Hello")).text);
} finally {
  await server.close();
}
from haarness_sdk import LocalServer

with LocalServer.start("/path/to/project") as server:
    thread = server.client.create_thread({})
    print(server.client.chat(thread["id"], "Hello")["text"])

If the folder's server is already running, for example one the terminal UI started, it is never replaced: your private server serves it at its own address, and closing yours leaves that server and its work running. Pass executable if haarness is not on your PATH, and disableProjectConfig: true (disable_project_config=True in Python) for a repository you do not trust, so its settings, hooks and MCP servers are not loaded.

If the server cannot start, the error says why in Haarness's own words, for example Haarness failed to start: … followed by the reason it printed.

Do anything else the server can

Everything the terminal UI and Desktop do goes through the same API, and each operation has a method named after it: createThread in TypeScript is create_thread in Python. Fields keep the server's JSON names, such as requestId and eventCursor, in both languages, and arguments go in the order path, body, query. Some you may want next:

  • readFile / writeFile (read_file / write_file) for the project's files. A write carries the revision you read, or NEW_FILE_REVISION for a new file, so it fails with 409 instead of overwriting someone's edit.
  • attachments() turns local images and PDFs into a turn's attachments.
  • btwStream (btw_stream) asks a side question without stopping the turn.
  • interruptTurn, listThreads, listItems and the chat's approval-mode you saw above.

sdks/OPERATIONS.md in the repository lists every method with its endpoint.

When a call takes long

Calls give up after 30 seconds unless you say otherwise. Give the whole client longer, or one call:

const client = new Haarness({
  baseUrl: "http://127.0.0.1:4096",
  timeoutMs: 60_000,
});
const info = await client.info({ timeoutMs: 5_000 });
client = Haarness("http://127.0.0.1:4096", timeout=60)
info = client.info(options={"timeout": 5})

To cancel a call, pass { signal } from an AbortController in TypeScript, or cancel the task in async Python. chat() already waits five minutes for its turn.

Event streams never time out on their own. A connection that died without closing (a laptop that slept, a network that changed) can look open for a long time, so tell the SDK how long silence may last: events({ idleTimeoutMs: 45_000, reconnect: true }) in TypeScript, or events(idle_timeout=45, reconnect=True) in Python. The server sends a keepalive every 15 seconds, so a healthy stream is never that quiet, and reconnect picks up again from the last event.

When something fails

A request the server refuses raises ApiError, with status, code, retryable and body; the codes are the API's. A connection problem or a timeout raises TransportError. The SDKs never retry on their own, and a turn is safe to send again with the same requestId: the server starts it only once.

import { randomUUID } from "node:crypto";
import { ApiError, TransportError } from "@haarness/sdk";

const turn = { requestId: randomUUID(), text: "Run the tests" };
try {
  await client.startTurn(thread.id, turn);
} catch (error) {
  if (error instanceof TransportError) await client.startTurn(thread.id, turn);
  else if (error instanceof ApiError)
    console.error(error.status, error.code, error.message);
  else throw error;
}
import uuid
from haarness_sdk import ApiError, TransportError

turn = {"requestId": str(uuid.uuid4()), "text": "Run the tests"}
try:
    client.start_turn(thread["id"], turn)
except TransportError:
    client.start_turn(thread["id"], turn)
except ApiError as error:
    print(error.status, error.code, error)