Build a calculator agent, then try the same loop through a stream. This guide uses the current Python API and a real model provider.

1. Install

Use Python 3.11 or newer in a virtual environment.

uv init my-agent && cd my-agent
uv add harnessx
export ANTHROPIC_API_KEY="your-api-key"

Any installer works; pip install harnessx gives the same package. Other model providers need their corresponding optional extra and configuration.

2. Create an agent

Save this as agent.py and run it with uv run agent.py.

import asyncio
from harnessx import Agent, AgentConfig, Limits

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

async def main():
    async with Agent(
        config=AgentConfig(
            system_prompt="Use multiply for multiplication. Be concise.",
            limits=Limits(max_iterations=10),
        ),
        tools=[multiply],
    ) as agent:
        result = await agent.run("What is 40 * 10?")
        if result.error:
            raise RuntimeError(result.error["message"])
        print(result.output)

asyncio.run(main())

python agent.py

The expected arithmetic is 400. The exact wording and number of model calls can vary. The tool is a normal local function; no expression evaluation or shell is involved.

3. Follow each event

For text only, agent.stream_text(prompt) yields the answer as it arrives and raises RunFailed if the run does not complete:

async def respond_text(agent, prompt):
    async for chunk in agent.stream_text(prompt):
        print(chunk, end="", flush=True)

To follow tool calls as well, replace the run call inside the agent context with an event stream:

from harnessx import RunEventType

async def respond(agent, prompt):
    text = ""
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.ATTEMPT_RESET:
                text = ""
            elif event.type == RunEventType.TEXT_DELTA:
                text += event.data
            elif event.type == RunEventType.TOOL_CALL_START:
                print("Tool:", event.data.name, event.data.input)
        result = await stream.result()
    if result.error:
        raise RuntimeError(result.error["message"])
    return result.output

Text deltas are provisional. A UI should clear an interrupted attempt on ATTEMPT_RESET and treat the final RunResult as authoritative. Closing the stream context cleans up an abandoned stream.

Try it without an API key

From a clone of the repository (uv sync --all-extras), these examples use scripted providers and real SDK behavior:

uv run python -m examples.skills_demo
uv run python -m examples.prompt_caching
uv run python -m examples.runtime_approvals
uv run python -m examples.flight_recorder --output incident.hx

They demonstrate skill invocation, prompt caching, persisted approvals, and offline playback. They are not live model benchmarks.

Next steps

  • Tools: register after construction or supply an explicit schema.
  • Permissions: require approval for selected effects.
  • Sessions: retain conversations and recover unfinished work.
  • Knowledge (OKF): give the agent a folder of curated facts to search and cite.
  • Web demo: configure prompts, skills, and MCP servers from a browser.