API reference - Plurai Simulation

This page lists exactly the 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.

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.

@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.

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 ToolMessages 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.

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.

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.

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.

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.

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.

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.

get_simulation_context

Returns the current SimulationContext.

ctx = plurai.get_simulation_context()

Returns SimulationContext.

world_state

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

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.

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.

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.

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.

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.

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.

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.

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.

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]`.

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.

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. Arguments (constructor, keyword-only)

simulate_path (Path): your harness.
settings: the SDK’s settings object.
observability (ObservabilitySink): the recording sink; see 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.

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().

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.

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.

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.

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.

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.

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.

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.

state = plurai.get_full_state()

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

get_events

Returns the recorded tool events of a session.

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.

plurai.emit_code_surface("plurai_sim/plurai.config.yaml", "plurai_sim/code_surface.yaml")

Version

version

The installed SDK version.

import plurai
print(plurai.__version__)   # "0.28.31"

plurai.observability

Client-side recording of Runs to one observability platform. See Trace sources for configuration.

create_observability_sink

Builds the one active sink from settings.

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.

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.

from plurai.observability import swallowable_error_types

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

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

​Writing simulate.py

​mock

​observe

​MockToolNode

​mock_wrap

​apply_mocks

​MockToolError

​mock_result

​configure

​SimulationContext

​get_simulation_context

​world_state

​world_state_snapshot

​active_session_id

​mock_service_url

​file_row

​store_record

​model_input_context

​HarnessFault

​AgentFault

​Host effects

​Running simulations

​Orchestrator

​OrchestratorClient

​ThreadResult

​OrchestratorAdapterError

​GraphRecursionWrap

​Tool and state contract

​ToolDescriptor

​Mock service

​MockServiceClient

​MockService

​start_session

​get_full_state

​get_events

​Code action channel

​Version

​__version__

​plurai.observability

​create_observability_sink

​ObservabilitySink

​NoOpSink

​LangSmithSink, LangfuseSink, ArizeSink, BraintrustSink

​LangSmithSinkConfig, LangfuseSinkConfig, ArizeSinkConfig, BraintrustSinkConfig

​LangSmithRunContext, LangfuseRunContext, ArizeRunContext, BraintrustRunContext

​RollupMetric

​swallowable_error_types