Rootmark SDK

Agent instrumentation for Rootmark — observe, verify and govern agent runs.

The SDK records what your agents actually did: the span tree of a run, the model calls inside it, token counts, and the payloads you choose to keep. It sends them to a Rootmark collector, which you can self-host.

pip install rootmark

Requires Python 3.12 or newer.

Quickstart

import rootmark

rootmark.init(
    api_key="pk_...",                 # project API key
    project_id="...",                 # UUID
    api_key_id="...",                 # UUID
    api_url="https://rootmark.example.com",
    workflow_name="my-service",       # not "default" -- see below
)

rootmark.patch()                      # instrument supported frameworks in place

That is the whole integration for the common case. patch() installs instrumentation process-wide, so call it once at startup and before the first agent invocation.

Grouping and labelling runs

with rootmark.workflow_override("contract-review"):
    with rootmark.bind_context(tenant_id="acme", session_id=session, tags=["surface:chat"]):
        result = agent(prompt)

bind_context applies to everything that runs inside it, including tasks the agent loop spawns, because the binding follows the async context.

workflow_name should not be "default": that value lets framework instrumentation override the name you bind per run with its own agent id.

Framework support

The base install instruments nothing on its own beyond direct use. Install the extra for what you actually run:

pip install "rootmark[openai]"    # OpenAI 1.x and 2.x
pip install "rootmark[langchain]" # LangChain callback handler
pip install "rootmark[crewai]"    # CrewAI event listeners
pip install "rootmark[strands]"   # Strands agents / AWS Bedrock
pip install "rootmark[boto3]"     # Bedrock via botocore, no code change

For boto3, a session built by hand rather than the default one needs rootmark.patch_session(session) — the botocore hooks only reach the default session otherwise.

Delivery

Spans are batched on a background thread and flushed on a timer, so nothing blocks a request. A failed export is written to a spill file and retried, keeping its batch identity so the collector can recognise a replay rather than double-count it. Ordering is preserved: a batch that cannot be delivered holds back the ones behind it.

Point spill_file somewhere writable. The default is relative to the working directory, which in a container is frequently owned by root while the process runs as someone else — and a spill file that cannot be written is buffering you silently do not have.

Call rootmark.flush() before a process exits if you want outstanding spans delivered, and rootmark.shutdown() to stop the processor.

Keeping payloads out

rootmark.init(..., suppress_payloads=True)

Spans are still recorded — timings, token counts, the shape of the run — but prompt and completion text is not sent. Use this when the collector sits outside the trust boundary that the data does not leave.

Links

• Documentation: https://rootmark.dev/docs/
• Source: https://github.com/IvanAdactic/rootmark