Python SDK API

The high-level SDK is exported from actionrail.

ActionRuntime

ActionRuntime(
    config=None,
    *,
    sources=None,
    ctx=None,
    monitor=False,
    on_decision=None,
    agent_id=None,
    api_key=None,
    endpoint="http://127.0.0.1:8020",
    review_timeout=300.0,
    review_poll=2.0,
    reviewer=None,
    state_dir=None,
    refresh_interval=30.0,
    max_config_staleness=86400.0,
    actions=None,
)

Framework-neutral runtime for explicit action evaluation and guarded execution. Supply either local config= or a Console agent_id=. Managed mode supports the same configuration refresh, Sources, reviews, durable reporting, monitor mode, and stale-configuration behavior as enforce().

Member	Purpose
evaluate(action, arguments=None, *, context=None)	Return a Decision without calling an operation.
aevaluate(...)	Run decision work without blocking the event loop.
execute(action, arguments, operation, *, context=None)	Evaluate, then call operation(**arguments) only when authorized.
aexecute(...)	Guard a sync or async operation from async application code.
register_actions(actions)	Replace and report a non-empty code-defined action surface.
report_surface(surface)	Advanced adapter hook for reporting an already-built, privacy-safe action-surface mapping in managed mode.
actions	Tuple of registered ActionDefinition objects.
health	Managed configuration and audit-delivery health.
flush(timeout=None)	Wait for current decision reports without closing.
close(timeout=None)	Stop refresh work and perform a bounded report flush.

The runtime is a context manager and calls close() on exit. Denied guarded operations raise ActionBlocked or ActionHeld, both subclasses of ActionNotAllowed with .action, .decision, and .review_status fields.

Ordinary applications should declare actions with ActionDefinition and register_actions(). report_surface() exists for framework-adapter authors that already produce ActionRail's action-surface mapping; it is a no-op without a managed reporter and must never contain runtime argument values.

See the direct Python integration.

ActionDefinition

ActionDefinition(
    name,
    arguments=(),
    description="",
    kind="consequential",
    critical_arguments=(),
)

Privacy-safe metadata for one explicitly registered action. Supported kinds are consequential, ingestion, memory, and neutral. Every critical argument must also appear in arguments. Definitions report names and classification only, never invocation values.

Decision

Decision.outcome is allow, hold, or block. The convenience properties allowed, held, and blocked provide explicit boolean checks. Local reasons and preview can contain sensitive values; report_* and model_* fields are separately privacy-reduced representations.

run_action_tests()

run_action_tests(
    config_path,
    cases_path,
    live_sources=False,
) -> ActionTestReport

Execute versioned action contracts through the production rule and decision pipeline. Configured Sources are not constructed by default; every referenced Source must have a deterministic case fixture. Set live_sources=True only in an explicit integration-test environment.

ActionTestReport.successful is true when every expected outcome matched. passed, failed, results, and to_dict() expose the complete result. Invalid documents raise ActionTestError; runtime failures remain attached to the affected result so later cases still execute.

See Action Tests and the schema reference.

enforce()

enforce(
    agent,
    config=None,
    sources=None,
    ctx=None,
    on_decision=None,
    agent_id=None,
    api_key=None,
    endpoint="http://127.0.0.1:8020",
    monitor=False,
    review_timeout=300.0,
    review_poll=2.0,
    reviewer=None,
    state_dir=None,
    refresh_interval=30.0,
    max_config_staleness=86400.0,
)

Wrap a compiled LangGraph agent and gate its tool entry points.

Parameter	Type	Description
agent	compiled graph	Agent whose discovered tools will be wrapped.
config	path or GroundConfig	Local rule configuration. Mutually exclusive primary mode with agent_id.
sources	dict[str, Source]	Optional locally constructed Source objects keyed by rule Source name.
ctx	mapping	Static trusted-context defaults. Prefer invocation-scoped trusted_context().
on_decision	callable	Local callback (tool_name, arguments, decision). Receives sensitive local detail.
agent_id	string	Registered Console agent UUID for managed configuration/reporting.
api_key	string	One-time agent key used for runtime control-plane endpoints.
endpoint	string	Control-plane base URL.
monitor	bool	Force evaluation/reporting without enforcing the outcome.
review_timeout	seconds	Maximum wait for a held action.
review_poll	seconds	Review status polling interval.
reviewer	object	Advanced custom reviewer implementing await_decision; primarily useful in tests/embedding.
state_dir	path	SDK configuration-cache and outbox root.
refresh_interval	seconds	Managed configuration refresh period; 0 disables background refresh.
max_config_staleness	seconds or None	Maximum unvalidated snapshot age before enforcement expires.

Provide either config= or agent_id=. Passing config, agent_id, and no sources intentionally allows local rules to resolve registered Source metadata from the Console.

The function returns the wrapped agent. Always assign the result:

agent = enforce(agent, agent_id=agent_id, api_key=agent_key)

Setup can raise for missing configuration mode, invalid policy, invalid Source construction, negative staleness, live configuration failure without a usable cache, or an over-stale enforcing cache.

instrument()

instrument(
    agent,
    agent_id,
    api_key,
    endpoint="http://127.0.0.1:8020",
    state_dir=None,
)

Report static action-surface discovery immediately and observed tool-call shapes after invoke(). instrument() never evaluates rules or blocks tools.

It returns the instrumented agent or a proxy. Always assign the result.

trusted_context()

trusted_context(values=None, /, **fields)

Context manager that binds authenticated invocation values through Python ContextVar:

with trusted_context({"workspace_id": "ws-1"}, customer_id="cus-1"):
    result = agent.invoke(inputs)

Nested scopes inherit and can override parent fields. The binding resets on normal or exceptional exit.

current_trusted_context()

current_trusted_context() -> dict

Return a copy of the context currently bound to this sync/async execution context. Mutating the returned dictionary does not change the active binding.

Reporting lifecycle

flush_reporting()

flush_reporting(agent, timeout=None) -> bool

Wait for current durable reports to be acknowledged without closing the reporter. Returns True when no events remain pending before the deadline.

close_reporting()

close_reporting(agent, timeout=None) -> bool

Stop configuration refresh, attempt a bounded report flush, and stop the reporter worker. Pending rows remain durable when the return value is False.

get_runtime_health()

get_runtime_health(agent) -> dict

Return top-level healthy, degraded, or inactive status with attached configuration and reporting component health.

Discovery helpers

The following are exported from actionrail.sdk (not the top-level actionrail) for advanced inspection and tooling — for example, from actionrail.sdk import enumerate_tools:

Function/type	Purpose
enumerate_tools(agent)	Statically discover tool name, kind, description, arguments, and suggested handling.
build_surface_map(agent, sample_inputs=None)	Build a SurfaceMap; optional samples add observed call counts/types.
observe_run(agent, sample_inputs, thread="scan")	Invoke sample prompts and collect call-shape metadata. This runs the agent and its tools.
calls_from_messages(messages)	Extract tool names and argument type shapes from LangChain AI messages.
print_surface_map(surface_map)	Print a human-readable discovery report.
ToolInfo	Dataclass for one discovered tool.
SurfaceMap	Dataclass containing store/memory metadata and discovered tools.

observe_run() executes the agent

Use safe sample inputs and non-production tools. Static discovery through enumerate_tools() or build_surface_map() without samples does not make model or tool calls.

Source classes

Import built-in implementations from actionrail.sdk.grounding:

from actionrail.sdk.grounding import (
    HTTPSource,
    MCPSource,
    MySQLSource,
    PostgresSource,
    SQLiteSource,
)

All implement the Source protocol:

class Source(Protocol):
    def ground(self, value, check, ctx, args=None) -> GroundVerdict:
        ...

GroundVerdict contains grounded: bool, a detailed local detail: str, and optional local matched data.