> ## Documentation Index
> Fetch the complete documentation index at: https://docs.plurai.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# API reference

> The supported Python surface of the plurai SDK: the names exported from the package root and from plurai.observability.

This page lists exactly the [stability contract](https://docs.plurai.ai/sdk/overview.md#stability-contract): the names in
`plurai.__all__` and in `plurai.observability.__all__`. Anything else you can import is implementation detail and
may change between versions.

## Writing simulate.py

### mock

Registers a mock handler for one tool, with the descriptor the backend uses to generate the mock environment.

```python theme={null}
import plurai

@plurai.mock("get_user_profile", descriptor={
    "description": "Fetch a user profile",
    "args": {"user_id": {"type": "str"}},
    "returns": {"type": "dict"},
})
def mock_get_user_profile(user_id: str):
    return plurai.get_simulation_context().world_state.get_entity(user_id)
```

**Returns** a decorator that registers the handler.

**Arguments**

* `tool_name` (`str`): the tool's name, as in `plurai.config.yaml`.
* `descriptor` (`dict | None`): the tool's descriptor. `description` and `args` are filled from the handler's
  docstring and signature when omitted.
* `input_schema`, `output_schema` (keyword): a Pydantic model for typed calls. Detected from a single
  Pydantic-typed parameter and a Pydantic return annotation when omitted; pass `None` to opt out.
* `**desc_kwargs`: descriptor fields as keyword arguments.

### observe

Registers an observer for a tool that runs for real, recording its result into the world.

```python theme={null}
@plurai.observe("rag_search")
def observe_rag_search(args, result):
    ctx = plurai.get_simulation_context()
    ctx.world_state.entities["search_result_0"] = {"_entity_type": "search_result", **result[0]}
```

**Returns** a decorator that registers the observer.

**Arguments**

* `tool_name` (`str`): the tool's name.
* `capture` (`Callable | None`): an optional capture function.

### MockToolNode

A drop-in replacement for a LangGraph `ToolNode` that routes each tool call by its mode.

```python theme={null}
node = plurai.MockToolNode(
    tools=[get_user_profile, process_refund],
    mock_config={"get_user_profile": "mock", "process_refund": "mock"},
)
graph.add_node("tools", node)
```

**Returns** a node with `invoke` and `ainvoke`, whose `ToolMessage`s match `ToolNode`'s (`name`, `tool_call_id`,
`status`, `artifact`, content). A `real` or `observe` result is rendered as JSON. A failed call is a message with
`status="error"`, not a raised exception. Calls run one after another, and injected state, store or runtime
arguments are not supplied.

**Arguments**

* `tools` (`list`): the agent's tools.
* `mock_config` (`dict[str, str] | None`): tool name to mode, `"mock"` (default), `"observe"` or `"real"`.
* `name` (`str`, keyword): the node name. Default `"tools"`.
* `messages_key` (`str`, keyword): the state key holding messages. Default `"messages"`.

### mock\_wrap

Wraps one tool callable so its calls route through the mock environment.

```python theme={null}
get_user_profile = plurai.mock_wrap(get_user_profile)
save_note = plurai.mock_wrap(save_note, failure=lambda e: {"error": e.message})
```

**Returns** the wrapped callable. It keeps the tool's `name`, `description` and `args_schema`.

**Arguments**

* `func` (`Callable`): the tool.
* `mode` (`str`): `"mock"` (default), `"observe"` or `"real"` (returns the tool unchanged).
* `failure` (`Callable[[MockToolError], Any] | None`): for a tool whose `error_channel` is `returns`, builds the
  failure value the real tool returns. A tool declared `raises` raises `MockToolError` instead.

### apply\_mocks

Wraps every named function on a module in place with `mock_wrap`.

```python theme={null}
import my_agent.tools as tools

patched = plurai.apply_mocks(tools, {"get_user_profile": "mock", "charge_card": "mock"})
```

**Returns** `list[str]`: the names actually patched. A tool built inside a factory closure or exported under
another name is not reachable and not patched.

**Arguments**

* `module`: the module that defines the tools.
* `mock_config` (`dict[str, str]`): tool name to mode.

### MockToolError

The exception a failed mocked call raises.

```python theme={null}
try:
    result = plurai.mock_result(response, channel="raises")
except plurai.MockToolError as err:
    return {"error": err.message}
```

**Attributes**

* `message` (`str`): the failure text.
* `case` (`str`): which kind of failure: `error_message`, `engine_refusal`, `handler_error` or
  `drifted_status_with_content`.
* `response`: the response as served.

### mock\_result

Turns a mock response into what your agent sees: the tool's value, or a raised `MockToolError`.

```python theme={null}
value = plurai.mock_result(response)
```

**Returns** the tool's value: `content` of a success, the error map of a failure the tool reports through its
value, or the response whole when it has no standard shape. Raises `MockToolError` for any other failure.

**Arguments**

* `response`: the mock response.
* `channel` (`"raises" | "returns" | None`): the tool's declared `error_channel`. `raises` turns a reported failure
  into a raised `MockToolError`.

### configure

Connects this process to a mock service, or hosts one in-process for local development.

```python theme={null}
plurai.configure("https://<your-backend>/mock/<env_id>/")
plurai.configure(mode="local")
```

**Returns** `None`. In `connect` mode it raises when the mock is unreachable or unhealthy; it never falls back to
a local mock.

**Arguments**

* `service_url` (`str | None`): the mock service URL. Optional in `local` mode, where the host and port are not
  honored.
* `mode` (`"connect" | "local"`, keyword): default `"connect"`.
* `mock_service_url` (`str | None`, keyword): an alias of `service_url`. Pass one of the two.

### SimulationContext

The scenario's context, available in mock handlers, replacement nodes and your own code.

```python theme={null}
ctx = plurai.get_simulation_context()
user_id = ctx.agent_context.get("user_id")
```

**Attributes**

* `agent_context` (`dict`): the scenario's context values.
* `world_state`: the world proxy. See [`world_state`](#world_state).

### get\_simulation\_context

Returns the current `SimulationContext`.

```python theme={null}
ctx = plurai.get_simulation_context()
```

**Returns** `SimulationContext`.

### world\_state

The scenario's world, read and written through the mock service.

```python theme={null}
user = plurai.world_state.get_entity("user_1")
refunds = plurai.world_state.query("transaction", user_id="user_1", type="refund")
plurai.world_state.apply({"user_1": {"balance": 40.0}})
```

**Methods**

* `get_entity(entity_id)`: one entity, or `None`.
* `query(entity_type, **filters)`: entities of a type matching exact field values.
* `apply(changes)`: write `{entity_id: {fields}}`. Each new entity carries `_entity_type`.
* `entities`: a mapping of every entity. It is a `Mapping`, not a `dict`.
* `generate_uuid()`: a new id.

### world\_state\_snapshot

Returns the whole world as a plain dict, in one call.

```python theme={null}
snapshot = plurai.world_state_snapshot()
by_type = {}
for entity in snapshot.values():
    by_type[entity["_entity_type"]] = by_type.get(entity["_entity_type"], 0) + 1
```

**Returns** `dict[str, dict]`: entity id to entity. Free inside a mock handler; one HTTP call elsewhere. `{}` when
no mock client is configured.

**Arguments**

* `session_id` (`str | None`): default the active session. Raises `RuntimeError` when none is active.

### active\_session\_id

Returns the id of the session active in this context.

```python theme={null}
store_key = f"{plurai.active_session_id()}:{record_id}"
```

**Returns** `str | None`. `None` outside a scenario.

### mock\_service\_url

Returns the mock service URL this process is configured against.

```python theme={null}
url = plurai.mock_service_url()
```

**Returns** `str | None`. `None` before `configure`.

### file\_row

The `world_snapshot()` row for one file your agent produced or was handed.

```python theme={null}
world["output_file:report.xlsx"] = plurai.file_row(out_dir / "report.xlsx", entity_type="output_file")
```

**Returns** `dict` with `_entity_type`, `name`, `suffix`, `size_bytes`, `sha256` and, for a previewed file,
`text_preview` (at most 4,000 characters, with `_preview_truncated` when cut). Spreadsheet formulas appear as their
formula text with the value the file saved; nothing is calculated. A file too large or too slow to read is
`_unreadable` with a reason.

**Arguments**

* `path` (`str | PathLike`): the file.
* `entity_type` (`str`, keyword): the row's type.
* `preview` (`bool`, keyword): `True` (default) for a file your agent produced. `False` for a file handed to it,
  which adds `content_type` instead of a preview.

### store\_record

The record of your harness's own store that a seeded world row stands for.

```python theme={null}
def seed_state(initial_agent_context, initial_world_state, resume_seed=None):
    for row in initial_world_state["entities"].values():
        my_store.insert(plurai.store_record(row))
```

**Returns** `dict`: the row without `_`-prefixed keys and without the backend's `id` copy. A row with no
`_entity_type`, or of an undeclared type, raises `HarnessFault`.

**Arguments**

* `row` (`Mapping`): a seeded row.
* `entity_types` (`Mapping | None`, keyword): default `world_state.entity_types` from the config beside the harness.

### model\_input\_context

The `input_context` entries for one turn: what your agent's model call received that the conversation does not
carry.

```python theme={null}
return {
    "messages": result["messages"],
    "input_context": plurai.model_input_context(model_input, conversation, source="memory_loader"),
}
```

**Returns** `list[dict[str, str]]`: `{"source", "content"}` per injected text, in model-input order. Agent-side
messages, a leading system prompt, blank text and text the conversation already carries are left out.

**Arguments**

* `model_input`: the messages the model call received.
* `conversation`: the messages the harness handed the agent.
* `source` (`str`, keyword): the read the entries came from.

### HarnessFault

Raised by `simulate.py` for a refusal the harness makes, so the scenario is not scored against your agent.

```python theme={null}
raise plurai.HarnessFault("refund_window node is unreachable from simulate.py")
```

Recorded as `not_valid`. Recorded when constructed, so an agent tool that catches it does not hide it.

### AgentFault

Raised by a harness on your agent's behalf, when the agent's own code failed where no Python frame shows it.

```python theme={null}
raise plurai.AgentFault(f"agent raised in the bridge child: {message}")
```

Raised from `invoke_agent`, recorded as `agent_raised`, a failed trial your agent owns. A provider failure and an
earlier `HarnessFault` in the turn still win.

### Host effects

For an agent built on `claude_agent_sdk`, these confine each scenario's Claude Code CLI session.

| Name | Purpose |
| - | - |
| `confine(options, *, run_dir, repo, workdir, trail)` | Confines one scenario's `claude_agent_sdk` options before connect. Returns the record the harness reports as `agent_state["host_effects"]`, and appends every tool call to `trail`. |
| `host_read_scope(*, run_dir, repo, harness=...)` | The read scope of the agent for one scenario: what it may and may not read. |
| `cli_hooks(scope, *, trail, on_read_denied=None)` | The hooks every confined session carries. Pass as `ClaudeAgentOptions(hooks=...)`. |
| `host_effects_granted(workdir)` | The tool names `<workdir>/.host_effects_consent` grants, as a `frozenset[str]`. |

```python theme={null}
record = plurai.confine(session.options, run_dir=run_dir, repo=REPO, workdir=HERE, trail=trail)
```

## Running simulations

These REST client types live in `plurai.orchestrator`, not the package root.

### Orchestrator

Drives one scenario end to end: opens the conversation, runs your harness in a child process, plays the turns, and
finalizes for scoring.

```python theme={null}
from plurai.orchestrator import Orchestrator

async with Orchestrator(
    simulate_path=Path("plurai_sim/simulate.py"),
    settings=settings,
    observability=sink,
    client=client,
) as orchestrator:
    result = await orchestrator.run(
        scenario_set_id="<id>",
        scenario_id="<id>",
        eval_set_id="<id>",
        redrive_available=False,
        allow_turn_timeout_redrive=False,
    )
```

**Returns** from `run`: a [`ThreadResult`](#threadresult).

**Arguments** (constructor, keyword-only)

* `simulate_path` (`Path`): your harness.
* `settings`: the SDK's settings object.
* `observability` (`ObservabilitySink`): the recording sink; see [`create_observability_sink`](#create_observability_sink).
  The orchestrator never closes it.
* `client` (`OrchestratorClient`): the backend client.
* `shape` (`"dialog" | "pipeline" | None`): a debugging override; `None` lets the bench's declaration decide.

**Arguments** (`run`, keyword-only): `scenario_set_id`, `scenario_id`, `eval_set_id`, `max_turns` (default 5),
`agent_version`, `mock_service_url`, `scenario_time_budget`, and the required `redrive_available` and
`allow_turn_timeout_redrive`.

### OrchestratorClient

The async client for the backend's simulation API.

```python theme={null}
client = plurai.orchestrator.OrchestratorClient("https://<your-backend>/v1", api_key=token)
```

**Arguments**

* `base_url` (`str`): the backend base URL.
* `timeout` (`float`): read timeout in seconds. Default 1200.
* `api_key` (`str | None`): the bearer token.
* `retry`: the retry policy. Default: the SDK's.

### ThreadResult

The outcome of one `Orchestrator.run()`.

```python theme={null}
print(result.turns, result.ended_early)
```

**Fields** include `thread_id`, `turns`, `finalize` (the scoring response), `user_messages`, `chatbot_messages`,
`tool_events`, `transcript_turns`, `ended_early` and `ended_early_cause`, and `resume_seed`. `summary()` returns a
short description.

### OrchestratorAdapterError

Raised when your harness's `invoke_agent`, `seed_state` or `invoke_pipeline` hits a misuse the runner can detect.

```python theme={null}
try:
    await orchestrator.run(...)
except plurai.orchestrator.OrchestratorAdapterError as err:
    print(err.child_exc_class, err.timed_out)
```

A `RuntimeError`. **Attributes**: `child_exc_class` (the original exception class in your harness),
`timed_out`, `thread_id`, `raised_at` (`"init"` or `"finalize"`). The original exception is `__cause__`.

### GraphRecursionWrap

An `OrchestratorAdapterError` raised when your LangGraph agent hit its recursion limit.

```python theme={null}
except plurai.GraphRecursionWrap as err:
    print(err.cause_class)
```

**Attributes**: `cause_class` (`"local_seed"`, `"remote_mock"` or `"unknown"`, best-effort), `thread_id`,
`turn_idx`. It counts as your agent's failed trial only when the agent was not silent before it crashed.

## Tool and state contract

### ToolDescriptor

The schema of one `tools.<name>.descriptor` block in `plurai.config.yaml`.

```python theme={null}
from plurai import ToolDescriptor

ToolDescriptor.model_validate({"description": "Fetch a user profile"})
```

**Fields** include `description` (required), `original_module`, `original_function`, `reads_from`, `writes_to`,
`stateful`, `args`, `returns`, `agent_usage` and `business_rules`. A descriptor whose `writes_to` is set must
declare post-conditions.

## Mock service

### MockServiceClient

The HTTP client for a mock service, local or remote.

```python theme={null}
client = plurai.MockServiceClient("https://<your-backend>/mock/<env_id>/")
client.health()
```

**Arguments**

* `base_url` (`str`): the mock service URL.

### MockService

The mock engine: world state, events and your registered handlers, served over HTTP. `configure(mode="local")`
builds one for you.

```python theme={null}
service = plurai.MockService(port=8199, host="127.0.0.1")
```

**Arguments**: `port` (default 8199), `host` (default `127.0.0.1`), and optional `handlers`, `observers`,
`descriptors`, `schemas`. `LocalMockService` is a deprecated alias.

### start\_session

Starts a simulation session against the configured mock service and makes it the active one.

```python theme={null}
plurai.start_session("sess-1", initial_world_state={"entities": {...}}, agent_context={"user_id": "u1"})
```

**Returns** the server's response (`dict`). Concurrent scenarios are isolated per `asyncio` task.

**Arguments**: `session_id` (`str`), `initial_world_state` (`dict`), `agent_context` (`dict | None`).

### get\_full\_state

Returns the current world state of a session.

```python theme={null}
state = plurai.get_full_state()
```

**Arguments**: `session_id` (`str | None`), default the active session.

### get\_events

Returns the recorded tool events of a session.

```python theme={null}
events = plurai.get_events()
```

**Arguments**: `session_id` (`str | None`), default the active session.

## Code action channel

For an agent that calls a vendor SDK or CLI from code it runs (for example a skill snippet), these route those
calls to the mock environment. The surface is projected from the mock tools in `plurai.config.yaml`.

| Name | Purpose |
| - | - |
| `build_code_surface(config)` | Projects a loaded `plurai.config.yaml` dict into a `code_surface.yaml` document. Pure; no I/O. |
| `emit_code_surface(config_path, out_path)` | Reads the config, builds the surface and writes it. Returns `out_path`. |
| `CodeSurfaceGenerationError` | Raised when the authored descriptors cannot be projected unambiguously. |
| `install_code_hook(code_surface_path, *, dispatch=...)` | Installs an import hook that shadows each vendor module the surface names. Returns the hook. |
| `install_code_hook_from_env(*, dispatch=...)` | The same, reading the surface path from `PLURAI_CODE_SURFACE`. Raises when it is unset. |
| `uninstall_code_hook(hook)` | Removes an installed hook. |
| `sitecustomize_source()` | The `sitecustomize.py` body that installs the hook at interpreter start. |
| `write_sitecustomize(target_dir)` | Writes that file into `target_dir` and returns its path. |
| `code_intercept_call(tool, args)` | Routes one intercepted call to the mock, using `PLURAI_SESSION_ID` and `PLURAI_MOCK_URL`. Fails closed when either is missing. |
| `watch_interception(*paths)` | Records the files your harness wrote for interception, so a Run whose agent rewrites them says so. |

```python theme={null}
plurai.emit_code_surface("plurai_sim/plurai.config.yaml", "plurai_sim/code_surface.yaml")
```

## Version

### \_\_version\_\_

The installed SDK version.

```python theme={null}
import plurai
print(plurai.__version__)   # "0.28.31"
```

## plurai.observability

Client-side recording of Runs to one observability platform. See
[Trace sources](https://docs.plurai.ai/integrations/trace-sources.md#record-runs) for configuration.

### create\_observability\_sink

Builds the one active sink from settings.

```python theme={null}
from plurai.observability import create_observability_sink

sink = create_observability_sink(settings)
```

**Returns** `ObservabilitySink`: the sink whose `<SINK>__ENABLED` flag is set, or a `NoOpSink` when none is.
Enabling more than one sink raises `ValueError` at startup.

**Arguments**: `settings`, the SDK's settings object.

### ObservabilitySink

The protocol every sink satisfies: tracing a Run and syncing its dataset. A malformed sink fails at startup.

### NoOpSink

The sink in use when none is configured. Every method does nothing.

### LangSmithSink, LangfuseSink, ArizeSink, BraintrustSink

The four sinks, each built from its config.

```python theme={null}
from plurai.observability import LangfuseSink, LangfuseSinkConfig

sink = LangfuseSink(LangfuseSinkConfig(enabled=True, public_key="pk", secret_key="sk"))
```

### LangSmithSinkConfig, LangfuseSinkConfig, ArizeSinkConfig, BraintrustSinkConfig

Each sink's settings, read from the matching `<SINK>__<FIELD>` environment variables.

| Config | Fields |
| - | - |
| `LangSmithSinkConfig` | `enabled`, `api_key`, `project`, `dataset`, `e2e_judge_enabled` |
| `LangfuseSinkConfig` | `enabled`, `public_key`, `secret_key`, `host`, `environment`, `project`, `dataset` |
| `ArizeSinkConfig` | `enabled`, `space_id`, `api_key`, `project_name`, `endpoint` |
| `BraintrustSinkConfig` | `enabled`, `api_key`, `project` |

### LangSmithRunContext, LangfuseRunContext, ArizeRunContext, BraintrustRunContext

The per-scenario context each sink's `prepare_run` returns and its later calls take. You do not build these
yourself.

### RollupMetric

One rollup feedback row a sink posts per scenario.

**Fields**: `key` (`str`), `score` (`float`), `comment` (`str`), `value` (`float | str`).

### swallowable\_error\_types

Builds the tuple of exception classes a sink may swallow, from `(module_path, class_name)` pairs that resolve.

```python theme={null}
from plurai.observability import swallowable_error_types

errors = swallowable_error_types(("httpx", "HTTPError"))
```

**Returns** `tuple[type[BaseException], ...]`, or `fallback` (default `(OSError,)`) when nothing resolves.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.