SDK

Running the agent

Execute tasks with run(), observe them with stream(), handle events, and cancel cleanly.

run()

run() executes the full agent loop and returns a RunResultonce the task finishes. It's the simplest way to get an answer or apply a change.

result = agent.run("Add pagination to GET /users")

result.text            # final assistant message
result.steps           # list[ToolStep]
result.files_changed   # ["src/routes/users.py"]
result.usage           # Usage(input_tokens=..., output_tokens=...)
result.session_id      # "sess_8f4ac1b"
result.success         # True

Messages and structured output

Pass a string for a single user turn, or a list of messages for finer control. Request structured output by passing a schema; the result's output is validated against it.

from pydantic import BaseModel

class Review(BaseModel):
    risk: str
    summary: str

result = agent.run(
    "Review the diff on this branch",
    output_schema=Review,
)
print(result.output.risk)   # typed, validated

stream()

stream() yields events as the agent works. Python returns a generator; TypeScript returns an AsyncIterable. The final event is always run_end, which carries the same RunResult you'd get from run().

stream = agent.stream("Migrate the config loader to TOML")

for event in stream:
    if event.type == "thinking":
        pass  # model reasoning, if exposed by the provider
    elif event.type == "text":
        print(event.text, end="", flush=True)
    elif event.type == "tool_call":
        print(f"\n⏺ {event.tool}", event.args)
    elif event.type == "tool_result":
        print("  ↳", event.status)
    elif event.type == "run_end":
        result = event.result

Event types

Every streamed event has a type discriminator:

Type	Payload	Emitted when
`run_start`	`session_id`	The loop begins.
`thinking`	`text`	The model emits reasoning (provider-dependent).
`text`	`text`	A chunk of the assistant's reply streams in.
`tool_call`	`tool`, `args`, `id`	The model invokes a tool.
`tool_result`	`id`, `status`, `output`	A tool finishes.
`permission_request`	`tool`, `args`	A call needs approval in `manual` mode.
`usage`	`input_tokens`, `output_tokens`	Token accounting updates.
`run_end`	`result`	The loop completes — final event.
`error`	`error`	A non-recoverable error ends the run.

Cancellation

Stop a run early without losing the partial session. Python accepts a cancel token or a KeyboardInterrupt; TypeScript accepts an AbortSignal.

from harnext import CancelToken

cancel = CancelToken()
# ... call cancel.cancel() from another thread / signal handler

result = agent.run("Long refactor", cancel=cancel)
if result.cancelled:
    print("stopped at step", len(result.steps))

Async Python

For asyncio, use await agent.arun(...) and async for event in agent.astream(...). The event shapes are identical to the synchronous API.

Running the agent

run()#

Messages and structured output#

stream()#

Event types#

Cancellation#

run()

Messages and structured output

stream()

Event types

Cancellation