# Quickstart

Build and run your first tool-using agent.

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.

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

Other model providers need their corresponding [optional extra and configuration](../providers/).

## 2. Create an agent

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

~~~python
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())
~~~

~~~bash
uv run agent.py
~~~

The expected arithmetic is **400**. The exact wording and number of model calls can vary.

## 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:

~~~python
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:

~~~python
from harnessx import RunEventType

async def respond(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
            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
~~~

Treat the final `RunResult` as authoritative, and closing the stream context cleans up an abandoned stream. Text deltas are provisional: a retry can restart the answer, which [Streaming](../streaming/) covers.

## Next steps

- [Research agent](../research-agent/): search the web and save a report with source links.
- [Tools](../tools/): register after construction or supply an explicit schema.
- [Permissions](../permissions/): require approval for selected effects.
- [Sessions](../sessions/): retain conversations and recover unfinished work.
- [Knowledge (OKF)](../knowledge/): give the agent a folder of curated facts to search and cite.
- [Web demo](../web-demo/): configure prompts, skills, and MCP servers from a browser.