Installing and Using the pycaber Auto SDK

Caber provides transparent, drop in automatic instrumentation for Python AI applications, web servers, HTTP and gRPC clients, WebSocket streams, and multiagent frameworks. Unlike passive observability tools, pycaber combines telemetry capture with synchronous in band policy enforcement, subdocument WRL Chunker characterization, On Behalf Of (OBO) Principal attribution, and surgical span redaction executed inside a compiled Rust core.

Figure 1: Five Stage In Process Interception, WRL Characterization, Edge Governance, and Output Rewriting Pipeline

## 1. Mixed Python and Rust Package Architecture (`v2.0.0`)

pycaber is built as a single mixed Python and Rust wheel using Maturin and PyO3 (0.22, abi3-py39). In packages/pycaber/pyproject.toml, Maturin is configured with python-source = "." and module-name = "pycaber._native".

2.2 Three Activation Patterns

• Mode 1: Single Line Import (import pycaber or import pycaber.auto): Place import pycaber at the top of your entry point before importing web frameworks, HTTP clients, or AI libraries.
• Mode 2: Zero Code CLI Wrapper (caber-run): Prefix any Python command with caber-run (packages/pycaber/pycaber/_cli.py). It injects packages/pycaber/pycaber/_bootstrap/sitecustomize.py onto PYTHONPATH so pycaber initializes automatically before application startup:

  CSI_CONFIG_PATH=/etc/caber/config.json \
    .venv/bin/caber-run .venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000

• Mode 3: Explicit Programmatic Configuration (pycaber.configure()): Call pycaber.configure(api_key=..., caber_host=..., redact_enabled=True, debug=False) to override configuration settings at runtime and log all loaded native and Python component versions (pycaber.get_version_info()).

3. HTTP, WebSocket, and gRPC Auto Instrumentation (15 Modules)

On import, pycaber registers lazy wrapt.when_imported() post import hooks across 15 client and server modules with zero startup overhead for unused libraries:

Category	Patched Module	Hook Target and Interception Behavior
HTTP Clients (Sync and Async)	requests	Wraps Session.send; extracts request and response bodies, injects outbound OBO and W3C trace headers, and applies synchronous response redaction.
HTTP Clients (Sync and Async)	httpx	Wraps Client.send and AsyncClient.send; supports buffered and streaming (aiter_bytes, iter_bytes) response interception and redaction.
HTTP Clients (Sync and Async)	aiohttp	Wraps ClientSession._request; intercepts async client calls and rewrites ClientResponse._body when redaction modifies content.
HTTP Clients (Sync and Async)	urllib3	Wraps HTTPConnectionPool.urlopen; includes a ContextVar recursion guard (_in_higher_level_patch) so calls already captured by requests are never double counted.
WebSocket Clients and Servers	websockets	Wraps sync and async connect, recv, and send for bidirectional frame capture, OBO context correlation, and frame redaction.
WebSocket Clients and Servers	websocket (websocket-client)	Wraps WebSocket.send, recv, recv_data, and create_connection.
WebSocket and Web Server	tornado	Wraps WebSocketHandler.on_message and write_message (including Streamlit Tornado sessions) plus AsyncHTTPClient.fetch.
ASGI and WSGI Web Frameworks	fastapi	Wraps FastAPI.__init__ to auto attach CaberASGIMiddleware for inbound HTTP and WebSocket governance.
ASGI and WSGI Web Frameworks	flask	Wraps Flask.__init__ to auto attach CaberWSGIMiddleware around app.wsgi_app.
ASGI and WSGI Web Frameworks	django	Wraps BaseHandler.get_response and get_response_async for sync and async Django views.
ASGI and WSGI Web Frameworks	bottle	Wraps Bottle.wsgi with CaberWSGIMiddleware.
ASGI and WSGI Web Frameworks	falcon	Wraps falcon.App.__init__ and falcon.asgi.App.__init__ for WSGI and ASGI apps.
ASGI and WSGI Web Frameworks	sanic	Wraps Sanic.handle_request across async request and response cycles.
ASGI and WSGI Web Frameworks	quart	Wraps Quart.asgi_app with CaberASGIMiddleware.
RPC Transports	grpc (grpcio)	Wraps insecure_channel, secure_channel, and grpc.aio channels with client interceptors across all four RPC cardinalities (unary_unary, unary_stream, stream_unary, stream_stream).

Selective Module Exclusion (CSI_GLOBAL_SDK_NOPATCH_MODULES)

To disable patching for specific libraries without turning off pycaber, set GLOBAL.sdk.noPatchModules in JSON configuration or pass a comma separated list via CSI_GLOBAL_SDK_NOPATCH_MODULES:

export CSI_GLOBAL_SDK_NOPATCH_MODULES="urllib3,tornado"

At runtime, call pycaber.get_patched_modules() to inspect active patches or pycaber.uninstrument() to restore all original unpatched functions cleanly.

4. In Process Agent Framework Governance (v2.10.0)

Modern agentic applications execute tool calls, RAG retrievals, handoffs, and state transitions entirely inside Python memory without making localhost HTTP requests. pycaber instruments 5 agent frameworks directly in process via packages/pycaber/pycaber/_frameworks/ and routes every action through the native Rust pycaber_process_http_exchange governance pipeline (packages/pycaber/pycaber/_agent_actions.py).

4.1 Supported Agent Frameworks and Patched Surfaces

Framework	Adapter Module	Patched Execution Surfaces
CrewAI	_frameworks/_crewai.py	Subscribes to crewai_event_bus (CrewKickoff*, AgentExecution*, Task*, LLMCall*, ToolUsage*, Memory*, Flow*, MCP*) and wraps Crew.kickoff, Crew.kickoff_async, Agent.execute_task, and BaseStructuredTool.invoke for inline output redaction.
LangChain	_frameworks/_langchain.py	Wraps Runnable.invoke, ainvoke, batch, abatch, BaseTool.invoke, ainvoke, BaseRetriever.invoke, ainvoke, and AgentExecutor._call, _acall.
LangGraph	_frameworks/_langgraph.py	Wraps Pregel.invoke, ainvoke, stream, astream (compiled StateGraph instances) and ToolNode._func, _afunc. Includes a _LANGGRAPH_ACTIVE ContextVar guard so internal LangChain Runnable steps inside a graph do not emit duplicate nested records.
LlamaIndex	_frameworks/_llamaindex.py	Registers an event handler on llama_index.core.instrumentation.get_dispatcher() (AgentRunStep*, AgentToolCall*, Workflow*, FunctionCall*, Retrieval*, Query*, Synthesize*, LLM*) and wraps BaseQueryEngine.query, aquery, BaseRetriever.retrieve, aretrieve, and BaseTool.__call__, acall.
AutoGen	_frameworks/_autogen.py	Wraps ConversableAgent.generate_reply, a_generate_reply, initiate_chat, a_initiate_chat, execute_function, a_execute_function (v0.2) and BaseChatAgent.on_messages, on_messages_stream, BaseGroupChat.run, run_stream, BaseTool.run_json (v0.4 autogen_agentchat and autogen_core).

4.2 Synthetic Request and Response Governance Contract

Each intercepted agent step or tool call is serialized as a synthetic HTTP exchange and evaluated against ApiMap host entries matching selfAC:<framework> (for example selfAC:langchain or selfAC:crewai) or the wildcard selfAC:*:

• Synthetic Request URL: http://selfAC:{framework}/{action_path} (with api_scheme="framework").
• MCP Tool Routing: When an agent invokes an MCP tool (detected via MCPToolWrapper, mcp server metadata, or MCP* events), action_path is normalized to mcp/{tool_name} so MCP tool policies and audit queries can target api.path: "mcp/*".
• OpenSearch csi_api_security Fields: Emits api.scheme: "framework", api.framework: "<framework>", capture.event_type: "<EventType>" (such as ToolInvoked, ToolReturned, RunnableCompleted, TaskCompletedEvent), event.code (200 on success, 500 on exception), and event.error.

4.3 Typed Return Value Rewriting (output_rewriter) and Fail Open Warning Contract

When the native Rust pipeline redacts restricted spans or PII from an agent step or tool return value, apply_redacted_output(result, original, *, rewriter=None) in _agent_actions.py applies the redacted JSON back to the Python return value:

1. Basic Python Types (str, bytes, dict, list, tuple): Replaced directly when every nested value is a basic JSON primitive (None, bool, int, float, str, bytes).
2. Registered output_rewriter Callbacks: For framework specific return types, adapters pass an explicit rewriter(original, redacted_value) hook. For example, LangChain tool executions often return a langchain_core.messages.ToolMessage object rather than a raw string. rewrite_tool_message_output (_frameworks/_langchain.py) constructs a shallow copy (original.model_copy() or copy.copy(original)) with its .content (and JSON serializable .artifact) replaced by the redacted output while preserving .tool_call_id, .name, .id, and .status so downstream LangGraph ToolNode consumers continue to function seamlessly.
3. Fail Open Refusal with Deduplicated Loud WARNING: When the native pipeline modifies the response (result.modified == True), pycaber checks whether the redacted value can be safely substituted. If the original return value is a complex object with no registered output_rewriter (such as CrewOutput, TaskResult, Response, or a LangGraph dict state containing live HumanMessage and AIMessage instances), substituting a plain JSON string or dict would crash the application with an AttributeError or TypeError. In that case, pycaber returns original unchanged (failing open for in place mutation while still recording the redacted audit event) and logs a loud WARNING deduplicated once per (framework, method, returned_type) tuple per process:

   pycaber: redaction modified output of langgraph:Pregel.invoke, but return value of type dict cannot be rewritten in place without breaking the caller; returning unredacted value. Register an output_rewriter for this wrapper to enable in-place redaction.

5. Framework Integration Recipes

5.1 FastAPI (Automatic ASGI Governance and OBO Header Extraction)

import pycaber  # Automatically hooks FastAPI.__init__ with CaberASGIMiddleware
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/v1/research")
async def research_endpoint(request: Request, body: dict):
    # Inbound X-User-Id / Authorization JWT and W3C traceparent are bound automatically
    # into InterceptContext and propagated to all outbound httpx/requests/agent calls
    return {"status": "ok", "query": body.get("query")}

5.2 Flask (Automatic WSGI Middleware and Custom User Header)

import pycaber  # Automatically wraps app.wsgi_app with CaberWSGIMiddleware
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/ask", methods=["POST"])
def ask():
    data = request.get_json()
    return jsonify({"reply": f"Processed for {request.headers.get('X-User-Id', 'anon')}"})

5.3 Streamlit (Per Session WebSocket Identity Pinning via _STREAMLIT_SESSION_REGISTRY)

Because Streamlit multiplexes all user interactions over Tornado WebSockets (/============================= or /_stcore/stream) and executes script reruns on worker threads decoupled from the Tornado I/O loop, pycaber maintains _STREAMLIT_SESSION_REGISTRY (packages/pycaber/pycaber/_patch_tornado.py). It bridges st.runtime.scriptrunner.get_script_run_ctx().session_id to the underlying Tornado WebSocketHandler connection key (ws://{host}/_stcore/stream#conn={id}):

import pycaber
import streamlit as st

# Pin the authenticated human user to the active Streamlit WebSocket session.
# Internally resolves session_id via _STREAMLIT_SESSION_REGISTRY even from Streamlit worker threads.
if "user_claims" in st.session_state:
    claims = st.session_state["user_claims"]
    pycaber.set_ws_user(
        subject=claims["sub"],
        issuer=claims["iss"],
        email=claims.get("email"),
        email_verified=bool(claims.get("email_verified", False)),
    )

prompt = st.chat_input("Ask the enterprise assistant")
if prompt:
    with st.chat_message("assistant"):
        st.write(run_rag_pipeline(prompt))
    # Signal turn boundary so buffered Streamlit protobuf frames flush as one logical response
    pycaber.mark_all_ws_responded()

5.4 LangChain and LangGraph (Tool Message Redaction and State Graph Governance)

import pycaber
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode

@tool
def fetch_patient_record(patient_id: str) -> str:
    """Fetches clinical summary governed by Caber OBO and HIPAA policies."""
    return f"Patient {patient_id}: SSN 123-45-6789, Diagnosis: Hypertension"

# ToolNode returns ToolMessage objects; pycaber automatically applies rewrite_tool_message_output
# so redacted content replaces ToolMessage.content while preserving tool_call_id and status.
tool_node = ToolNode([fetch_patient_record])

5.5 CrewAI, AutoGen, LlamaIndex, and MCP Tool Servers

import pycaber  # Patches CrewAI event bus, AutoGen agents, LlamaIndex dispatcher, and MCP tools
from crewai import Agent, Task, Crew

analyst = Agent(
    role="Financial Analyst",
    goal="Summarize quarterly earnings",
    backstory="Senior equity researcher",
)
task = Task(
    description="Analyze Q3 revenue guidance using MCP financial tools",
    expected_output="Sanitized summary",
    agent=analyst,
)
crew = Crew(agents=[analyst], tasks=[task])
# Crew.kickoff is governed under selfAC:crewai; any MCP tool invocation is logged under api.path="mcp/<tool_name>"
result = crew.kickoff()

6. Correlated WebSocket Interaction Registry (InteractionRegistry)

In streaming WebSocket architectures where application logic assembles high level logical turns across multiple raw frames, operators can enable Registered Content Mode on a host in ApiMap by setting registeredContentMode: true (or interactionMode: "registered").

When registeredContentMode is active on a HostConfig:

1. Raw WebSocket frame capture in _patch_tornado.py, _patch_websockets.py, and _patch_websocket_client.py suppresses automatic per frame ws_on_message dispatch.
2. The application explicitly demarcates logical request and response turns using pycaber.register_request and pycaber.register_response, backed by InteractionRegistry and BufferedMessage in pycaber:

import pycaber

# 1. Register logical user request and receive an interaction correlation handle
interaction_handle = pycaber.register_request(
    user_prompt_text,
    api="ws://localhost:8501/_stcore/stream",
)

# 2. Generate streaming chunks or multi step agent response
assistant_reply = generate_complete_turn(user_prompt_text)

# 3. Correlate response with the request handle; triggers WRL characterization,
#    OBO policy evaluation, and emits a single correlated csi_api_security record
pycaber.register_response(
    assistant_reply,
    interaction=interaction_handle,
)

7. Protobuf Handler Registration and Automatic gRPC Decoding

Many streaming UIs (such as Streamlit BackMsg and ForwardMsg frames) and gRPC microservices exchange binary Protocol Buffers payloads. pycaber decodes binary protobuf payloads into structured JSON, routes them through the native Rust characterization and OBO redaction pipeline under synthetic MIME type application/vnd.caber.<handler>.protobuf+json, and re encodes the redacted JSON back into valid binary wire format.

7.1 Four ApiMap.<host>.protobufHandler Modes

protobufHandler Value	Resolution Behavior
"" (empty, default)	Binary protobuf payloads are treated as opaque binary bytes and are not decoded.
"streamlit"	Built in handler (packages/pycaber/pycaber/_streamlit_proto.py) that decodes Streamlit BackMsg (c2s) and ForwardMsg (s2c) WebSocket frames and re encodes redacted ForwardMsg deltas.
"<custom_name>"	Looks up a custom Python codec registered via pycaber.proto_register_handler(name, decode_fn, encode_fn).
"auto"	Automatic gRPC descriptor extraction (packages/pycaber/pycaber/_grpc_auto_proto.py). Lazily queries google.protobuf.descriptor_pool.Default().FindMethodByName() using the gRPC /package.Service/Method path, builds MessageFactory classes on the fly, converts messages via MessageToJson and Parse, and caches method handlers in _METHOD_HANDLER_CACHE.

7.2 Registering a Custom Protobuf Handler (proto_register_handler)

import json
import pycaber
from google.protobuf.json_format import MessageToDict, ParseDict
from my_org_proto import chat_pb2

def decode_chat_frame(raw_bytes: bytes, direction: str) -> str | None:
    msg = chat_pb2.ChatRequest() if direction == "c2s" else chat_pb2.ChatResponse()
    msg.ParseFromString(raw_bytes)
    return json.dumps(MessageToDict(msg, preserving_proto_field_name=True))

def encode_chat_frame(redacted_json: str, original_bytes: bytes, direction: str) -> bytes | None:
    msg = chat_pb2.ChatRequest() if direction == "c2s" else chat_pb2.ChatResponse()
    ParseDict(json.loads(redacted_json), msg, ignore_unknown_fields=True)
    return msg.SerializeToString()

# Register handler referenced by ApiMap.<host>.protobufHandler = "chat_proto"
pycaber.proto_register_handler("chat_proto", decode_chat_frame, encode_chat_frame)