Agent

The high-level entry point for most use cases. Agent and AsyncAgent compose a Harness, SessionCache, and optional tools from a single configuration.

---

Agent

data_harness.Agent

Agent(
    system: str,
    *,
    adapter: ProviderAdapter | None = None,
    model: str | None = None,
    max_turns: int = 25,
    cache: SessionCache | None = None,
    run_dir: str | Path | None = None,
    execution: str = "inprocess",
    sandbox_options: dict[str, Any] | None = None,
    on_code: Callable[[str], Any] | None = None,
    code_only: bool = False,
    hooks: HookRegistry | None = None,
)

Bases: _AgentBase

High-level synchronous agent.

Agent composes a Harness, a SessionCache, and optional tools from a single configuration. Each call to run builds a fresh Harness with a fresh message history. Use session when you need multi-turn conversation state to persist across questions.

Example::

from data_harness import Agent

agent = Agent(system="You are a data analyst.", model="claude-sonnet-4-6")
print(agent.run("Compute the mean of [1, 2, 3]."))

model resolves an adapter the same way resolve_adapter does: a provider/model id routes to OpenRouter, deepseek-* to DeepSeek direct, otherwise Anthropic or OpenAI by name. Pass adapter= instead for a pre-built or custom adapter — it always wins over model::

from data_harness.llm.providers.anthropic import AnthropicAdapter

agent = Agent(
    system="You are a data analyst.",
    adapter=AnthropicAdapter(model="claude-sonnet-4-6", max_tokens=8192),
)

Parameters:

Name	Type	Description	Default
system	str	System prompt passed unchanged to every Harness run.	required
adapter	ProviderAdapter | None	Synchronous provider adapter. Takes priority over model.	None
model	str | None	Model id resolved into an adapter via resolve_adapter, when adapter is not given. With neither, resolution falls back to whichever provider API key is set in the environment.	None
max_turns	int	Hard cap on provider turns per run call.	25
cache	SessionCache | None	Shared SessionCache. A fresh cache is created when None.	None
run_dir	str | Path | None	Directory for chart artefacts and subagent working state. Defaults to ./runs.	None
execution	str	"inprocess" or "subprocess" interpreter isolation.	'inprocess'
sandbox_options	dict[str, Any] | None	Extra options for the subprocess interpreter.	None
on_code	Callable[[str], Any] | None	Approval gate called with interpreter code before it runs.	None
code_only	bool	When True, interpreter code is echoed, never executed.	False

Source code in data_harness/app/agent.py

def __init__(
    self,
    system: str,
    *,
    adapter: ProviderAdapter | None = None,
    model: str | None = None,
    max_turns: int = 25,
    cache: SessionCache | None = None,
    run_dir: str | Path | None = None,
    execution: str = "inprocess",
    sandbox_options: dict[str, Any] | None = None,
    on_code: Callable[[str], Any] | None = None,
    code_only: bool = False,
    hooks: HookRegistry | None = None,
) -> None:
    super().__init__(
        system,
        max_turns=max_turns,
        cache=cache,
        run_dir=run_dir,
        execution=execution,
        sandbox_options=sandbox_options,
        on_code=on_code,
        code_only=code_only,
        hooks=hooks,
    )
    self._adapter = adapter if adapter is not None else self._default_adapter(model)
    self._last_harness: Harness | None = None

session

session() -> AgentSession

Create a stateful AgentSession for multi-turn conversations.

Returns:

Type	Description
AgentSession	A new AgentSession backed by a copy of this agent's cache.

Source code in data_harness/app/agent.py

def session(self) -> AgentSession:
    """Create a stateful `AgentSession` for multi-turn conversations.

    Returns:
        A new `AgentSession` backed by a copy of this agent's cache.
    """
    return AgentSession(self)

run_result

run_result(user_message: str) -> RunResult

Run the agent and return the full RunResult.

Builds a fresh Harness with a fresh message history for each call.

Parameters:

Name	Type	Description	Default
user_message	str	The user prompt to send.	required

Returns:

Type	Description
RunResult	A RunResult with the text response, token usage, and cache state.

Source code in data_harness/app/agent.py

def run_result(self, user_message: str) -> RunResult:
    """Run the agent and return the full `RunResult`.

    Builds a fresh `Harness` with a fresh message history for each call.

    Args:
        user_message: The user prompt to send.

    Returns:
        A `RunResult` with the text response, token usage, and cache state.
    """
    key = self._replay_key(user_message)
    if key is not None:
        cached = self._exec_cache.get(key)
        if cached is not None:
            return self._replay(cached)

    harness = self._make_harness()
    self._last_harness = harness
    result = harness.run_result(
        user_message, run_id=str(uuid.uuid4()), session_id=None
    )
    self._last_result = result
    self._record_replay(key, harness, result)
    return result

run

run(user_message: str) -> str

Run the agent and return the final text response.

Parameters:

Name	Type	Description	Default
user_message	str	The user prompt to send.	required

Returns:

Type	Description
str	The model's final text response.

Raises:

Type	Description
MaxTurnsExceeded	If the loop reaches max_turns.
RuntimeError	If the provider raises an exception.

Source code in data_harness/app/agent.py

def run(self, user_message: str) -> str:
    """Run the agent and return the final text response.

    Args:
        user_message: The user prompt to send.

    Returns:
        The model's final text response.

    Raises:
        MaxTurnsExceeded: If the loop reaches ``max_turns``.
        RuntimeError: If the provider raises an exception.
    """
    return unwrap_text(self.run_result(user_message))

---

AgentSession

data_harness.AgentSession

AgentSession(agent: _AgentBase)

Bases: _SessionBase

Stateful chat session built from an Agent definition.

Agent.run() intentionally stays one-shot for examples and tests. Use Agent.session() when an application needs follow-up questions over the same message history and cache handles.

Source code in data_harness/app/agent.py

def __init__(self, agent: _AgentBase) -> None:
    self._agent = agent
    self._cache = SessionCache(
        sample_size=agent.cache.sample_size,
        storage_dir=None,
        hot_limit=agent.cache.hot_limit,
    )
    for name, value in agent.cache.items():
        self._cache.put(
            name,
            _copy_cache_value(value),
            semantics=agent.cache.get_semantics(name),
        )
    self._harness = agent._make_harness(cache=self._cache)  # type: ignore[attr-defined]
    self._id: str = str(uuid.uuid4())
    self._last_result: RunResult | None = None
    self._turns: int = 0

ask_result

ask_result(user_message: str) -> RunResult

Send a follow-up message and return the full RunResult.

Parameters:

Name	Type	Description	Default
user_message	str	The follow-up user prompt.	required

Returns:

Type	Description
RunResult	A RunResult for this turn sequence.

Source code in data_harness/app/agent.py

def ask_result(self, user_message: str) -> RunResult:
    """Send a follow-up message and return the full `RunResult`.

    Args:
        user_message: The follow-up user prompt.

    Returns:
        A `RunResult` for this turn sequence.
    """
    return self._record(
        self._harness.ask_result(
            user_message, run_id=str(uuid.uuid4()), session_id=self._id
        )
    )

ask

ask(user_message: str) -> str

Send a follow-up message and return the final text response.

Parameters:

Name	Type	Description	Default
user_message	str	The follow-up user prompt.	required

Returns:

Type	Description
str	The model's final text response.

Raises:

Type	Description
MaxTurnsExceeded	If the loop reaches max_turns.
RuntimeError	If the provider raises an exception.

Source code in data_harness/app/agent.py

def ask(self, user_message: str) -> str:
    """Send a follow-up message and return the final text response.

    Args:
        user_message: The follow-up user prompt.

    Returns:
        The model's final text response.

    Raises:
        MaxTurnsExceeded: If the loop reaches ``max_turns``.
        RuntimeError: If the provider raises an exception.
    """
    return unwrap_text(self.ask_result(user_message))

---

AsyncAgent

data_harness.AsyncAgent

AsyncAgent(
    system: str,
    *,
    adapter: AsyncProviderAdapter | None = None,
    model: str | None = None,
    max_turns: int = 25,
    cache: SessionCache | None = None,
    run_dir: str | Path | None = None,
    execution: str = "inprocess",
    sandbox_options: dict[str, Any] | None = None,
    on_code: Callable[[str], Any] | None = None,
    code_only: bool = False,
    hooks: HookRegistry | None = None,
)

Bases: _AgentBase

Async agent for use with AsyncProviderAdapter.

run() and run_result() are coroutines. run_stream() is an async generator that yields StreamEvents as they arrive from the provider. Use async_session() for multi-turn streaming conversations.

Takes the same configuration and supports the same features as Agent: connectors, MCP servers, SQL, the planner, subagents, the replay cache, and the interpreter approval gate. Also takes the same model= shortcut — see Agent for details; adapter= here must be an AsyncProviderAdapter.

Parameters:

Name	Type	Description	Default
system	str	System prompt passed unchanged to every AsyncHarness run.	required
adapter	AsyncProviderAdapter | None	Asynchronous provider adapter. Takes priority over model.	None
model	str | None	Model id resolved into an AsyncProviderAdapter via resolve_async_adapter, when adapter is not given. With neither, resolution falls back to whichever provider API key is set in the environment.	None
max_turns	int	Hard cap on provider turns per run call.	25
cache	SessionCache | None	Shared SessionCache. A fresh cache is created when None.	None
run_dir	str | Path | None	Directory for chart artefacts and subagent working state. Defaults to ./runs.	None
execution	str	"inprocess" or "subprocess" interpreter isolation.	'inprocess'
sandbox_options	dict[str, Any] | None	Extra options for the subprocess interpreter.	None
on_code	Callable[[str], Any] | None	Approval gate called with interpreter code before it runs.	None
code_only	bool	When True, interpreter code is echoed, never executed.	False

Source code in data_harness/app/agent.py

def __init__(
    self,
    system: str,
    *,
    adapter: AsyncProviderAdapter | None = None,
    model: str | None = None,
    max_turns: int = 25,
    cache: SessionCache | None = None,
    run_dir: str | Path | None = None,
    execution: str = "inprocess",
    sandbox_options: dict[str, Any] | None = None,
    on_code: Callable[[str], Any] | None = None,
    code_only: bool = False,
    hooks: HookRegistry | None = None,
) -> None:
    super().__init__(
        system,
        max_turns=max_turns,
        cache=cache,
        run_dir=run_dir,
        execution=execution,
        sandbox_options=sandbox_options,
        on_code=on_code,
        code_only=code_only,
        hooks=hooks,
    )
    self._adapter = adapter if adapter is not None else self._default_adapter(model)
    self._last_harness: AsyncHarness | None = None

async_session

async_session() -> AsyncAgentSession

Create a stateful AsyncAgentSession for multi-turn conversations.

Source code in data_harness/app/agent.py

def async_session(self) -> AsyncAgentSession:
    """Create a stateful `AsyncAgentSession` for multi-turn conversations."""
    return AsyncAgentSession(self)

run_result async

run_result(user_message: str) -> RunResult

Run the agent and return the full RunResult.

Source code in data_harness/app/agent.py

async def run_result(self, user_message: str) -> RunResult:
    """Run the agent and return the full `RunResult`."""
    key = self._replay_key(user_message)
    if key is not None:
        cached = self._exec_cache.get(key)
        if cached is not None:
            return await self._replay(cached)

    harness = self._make_harness()
    self._last_harness = harness
    result = await harness.run_result(
        user_message, run_id=str(uuid.uuid4()), session_id=None
    )
    self._last_result = result
    self._record_replay(key, harness, result)
    return result

run async

run(user_message: str) -> str

Run the agent and return the final text response.

Raises:

Type	Description
MaxTurnsExceeded	If the loop reaches max_turns.
RuntimeError	If the provider raises an exception.

Source code in data_harness/app/agent.py

async def run(self, user_message: str) -> str:
    """Run the agent and return the final text response.

    Raises:
        MaxTurnsExceeded: If the loop reaches ``max_turns``.
        RuntimeError: If the provider raises an exception.
    """
    return unwrap_text(await self.run_result(user_message))

run_stream async

run_stream(
    user_message: str,
) -> AsyncGenerator[StreamEvent, None]

Stream events for a one-shot run.

Yields StreamEvent objects (message_start, content_block_*, message_delta, message_stop, tool_result) following the Claude Agent SDK protocol.

After the generator is exhausted, agent.last_result holds the RunResult for the run, including token usage. Read it there rather than through last_harness: a replay-cache hit builds no harness, so last_harness is either None or left over from an earlier run.

A cache hit yields the cached answer as a single synthetic text block and reports zero usage, so a streaming caller sees the same answer a non-streaming one would.

Usage::

async for event in agent.run_stream("hello"):
    if event.type == "content_block_delta":
        from data_harness.llm.streaming import TextDelta
        if isinstance(event.delta, TextDelta):
            print(event.delta.text, end="", flush=True)

Source code in data_harness/app/agent.py

async def run_stream(self, user_message: str) -> AsyncGenerator[StreamEvent, None]:
    """Stream events for a one-shot run.

    Yields StreamEvent objects (message_start, content_block_*, message_delta,
    message_stop, tool_result) following the Claude Agent SDK protocol.

    After the generator is exhausted, ``agent.last_result`` holds the
    `RunResult` for the run, including token usage. Read it there rather
    than through ``last_harness``: a replay-cache hit builds no harness, so
    ``last_harness`` is either ``None`` or left over from an earlier run.

    A cache hit yields the cached answer as a single synthetic text block
    and reports zero usage, so a streaming caller sees the same answer a
    non-streaming one would.

    Usage::

        async for event in agent.run_stream("hello"):
            if event.type == "content_block_delta":
                from data_harness.llm.streaming import TextDelta
                if isinstance(event.delta, TextDelta):
                    print(event.delta.text, end="", flush=True)
    """
    key = self._replay_key(user_message)
    if key is not None:
        cached = self._exec_cache.get(key)
        if cached is not None:
            # `_replay` mints its own run id; nothing here to stamp.
            result = await self._replay(cached)
            for event in _synthetic_text_events(result.text):
                yield event
            return

    run_id = str(uuid.uuid4())
    harness = self._make_harness()
    self._last_harness = harness
    # `aclosing`, not a bare `async for`: closing this generator does not
    # close the one it iterates, so abandoning the stream here would leave
    # the harness (and the provider's HTTP stream) suspended.
    async with aclosing(harness.run_stream(user_message)) as events:
        async for event in events:
            yield event
    if harness.last_result is not None:
        self._last_result = harness._stamp(harness.last_result, run_id, None)
        self._record_replay(key, harness, harness.last_result)

---

AsyncAgentSession

data_harness.AsyncAgentSession

AsyncAgentSession(agent: _AgentBase)

Bases: _SessionBase

Stateful async chat session built from an AsyncAgent definition.

Source code in data_harness/app/agent.py

def __init__(self, agent: _AgentBase) -> None:
    self._agent = agent
    self._cache = SessionCache(
        sample_size=agent.cache.sample_size,
        storage_dir=None,
        hot_limit=agent.cache.hot_limit,
    )
    for name, value in agent.cache.items():
        self._cache.put(
            name,
            _copy_cache_value(value),
            semantics=agent.cache.get_semantics(name),
        )
    self._harness = agent._make_harness(cache=self._cache)  # type: ignore[attr-defined]
    self._id: str = str(uuid.uuid4())
    self._last_result: RunResult | None = None
    self._turns: int = 0

ask_result async

ask_result(user_message: str) -> RunResult

Send a follow-up message and return the full RunResult.

Source code in data_harness/app/agent.py

async def ask_result(self, user_message: str) -> RunResult:
    """Send a follow-up message and return the full `RunResult`."""
    return self._record(
        await self._harness.ask_result(
            user_message, run_id=str(uuid.uuid4()), session_id=self._id
        )
    )

ask async

ask(user_message: str) -> str

Send a follow-up message and return the final text response.

Raises:

Type	Description
MaxTurnsExceeded	If the loop reaches max_turns.
RuntimeError	If the provider raises an exception.

Source code in data_harness/app/agent.py

async def ask(self, user_message: str) -> str:
    """Send a follow-up message and return the final text response.

    Raises:
        MaxTurnsExceeded: If the loop reaches ``max_turns``.
        RuntimeError: If the provider raises an exception.
    """
    return unwrap_text(await self.ask_result(user_message))

ask_stream async

ask_stream(
    user_message: str,
) -> AsyncGenerator[StreamEvent, None]

Stream events for a follow-up turn.

After the generator is exhausted, session.last_result holds the RunResult for the turn, stamped with the session and run ids exactly as a non-streamed turn would be.

Source code in data_harness/app/agent.py

async def ask_stream(self, user_message: str) -> AsyncGenerator[StreamEvent, None]:
    """Stream events for a follow-up turn.

    After the generator is exhausted, ``session.last_result`` holds the
    `RunResult` for the turn, stamped with the session and run ids exactly
    as a non-streamed turn would be.
    """
    run_id = str(uuid.uuid4())
    async with aclosing(self._harness.ask_stream(user_message)) as events:
        async for event in events:
            yield event
    result = self._harness.last_result
    if result is not None:
        self._record(self._harness._stamp(result, run_id, self._id))
    else:  # pragma: no cover - the loop always sets a result
        self._agent._last_harness = self._harness

---

RunResult

data_harness.RunResult dataclass

RunResult(
    text: str,
    status: Literal[
        "success", "max_turns_exceeded", "error"
    ],
    turns: int,
    stop_reason: StopReason | None,
    usage: Usage,
    cache_snapshots: dict[str, str] = dict(),
    cache_storage: dict[str, CacheStorageInfo] = dict(),
    error: str | None = None,
    stopped_by: str | None = None,
    run_id: str | None = None,
    session_id: str | None = None,
    value: Any = None,
    charts: list[ChartArtifact] = list(),
)

The complete outcome of a single Harness.run() or Agent.run() call.

Attributes:

Name	Type	Description
text	str	The final text response from the model.
status	Literal['success', 'max_turns_exceeded', 'error']	"success", "max_turns_exceeded", or "error".
turns	int	Number of provider turns executed.
stop_reason	StopReason | None	Provider stop reason from the final turn, or None on error/max-turns.
usage	Usage	Cumulative token counts across all turns.
cache_snapshots	dict[str, str]	Mapping of handle name → compact snapshot string for every value in the session cache at the end of the run.
cache_storage	dict[str, CacheStorageInfo]	Mapping of handle name → CacheStorageInfo describing where each handle is stored.
error	str | None	Exception repr when status == "error", otherwise None.
stopped_by	str | None	Why a hook ended the run early, if one did. A capped run is otherwise indistinguishable from a model that answered with nothing.
run_id	str | None	Optional UUID assigned by Agent; None when using Harness directly.
session_id	str | None	Optional session UUID when the run is part of an AgentSession; None for one-shot runs.
value	Any	The structured final answer the model recorded via answer(...) inside the interpreter (a scalar, DataFrame, etc.), or None if it only produced prose.
charts	list[ChartArtifact]	Chart artefacts rendered during the run. Image bytes live on disk; only paths are referenced here.

---

Usage

data_harness.Usage dataclass

Usage(
    input_tokens: int = 0,
    output_tokens: int = 0,
    cache_read_tokens: int = 0,
    cache_write_tokens: int = 0,
)

Accumulated token counts for a run or session turn.

Attributes:

Name	Type	Description
input_tokens	int	Tokens in the prompt sent to the provider.
output_tokens	int	Tokens in the provider's response.
cache_read_tokens	int	Prompt tokens served from the provider's KV cache.
cache_write_tokens	int	Prompt tokens written to the provider's KV cache.

---

CacheStorageInfo

data_harness.CacheStorageInfo dataclass

CacheStorageInfo(
    location: Literal["memory", "disk"], storage_type: str
)

Where a named handle is physically stored in the SessionCache.

Attributes:

Name	Type	Description
location	Literal['memory', 'disk']	Either "memory" (hot) or "disk" (spilled cold).
storage_type	str	Format used on disk, e.g. "dataframe_parquet" or "numpy_npy". Always "memory" for hot entries.