Caber Systems Documentation

Appearance

pycaber SDK API Reference ​

This page provides the micro level technical reference for public functions, classes, decorators, and exceptions exposed by the pycaber Python package and its transparent instrumentation engine pycaber.auto.

pycaber SDK Architecture
Figure 1: pycaber SDK Framework Integration, Interception Engine, and Native Telemetry Bridge
## 1. Core Configuration & Native Lifecycle

configure ​

Configures pycaber and logs all native and Python component versions.

  • Syntax:
    python
    pycaber.configure(*args, **kwargs) -> None
  • Parameters:
    • *args: Positional arguments passed to internal configuration loaders.
    • **kwargs: Key value configuration overrides. Accepted keys include api_key (string), caber_host (string), redact_enabled (boolean), and debug (boolean).
  • Return Value: None.
  • Behavior & Environment: Initializes the native Rust telemetry engine, establishes HTTP or Unix domain socket transport to caber_core, reads CSI_API_KEY and CSI_CABER_HOST environment variables, and configures fallback logging. Must be invoked before importing user modules when running outside of caber-run.

uninstrument ​

Removes all Caber runtime monkey patches and restores original library functions.

  • Syntax:
    python
    pycaber.uninstrument() -> None
  • Parameters: None.
  • Return Value: None.
  • Behavior: Restores original unpatched functions across requests, httpx, urllib3, aiohttp, psycopg, and redis. Cleanly disables pycaber.auto without restarting the process.

get_patched_modules ​

Returns the set of module names currently monkey patched by pycaber.auto.

  • Syntax:
    python
    pycaber.get_patched_modules() -> set[str]
  • Parameters: None.
  • Return Value:
    • set[str]: Set containing active patched module names (for example {"requests", "httpx", "urllib3"}).

native_abi_report ​

Audits capability and binary compatibility of the native Rust extension exports.

  • Syntax:
    python
    pycaber.native_abi_report(module: Any = None) -> dict[str, AbiVerdict]
  • Parameters:
    • module (Any, optional, default: None): Optional module reference to audit. Defaults to internal native extension.
  • Return Value:
    • dict[str, AbiVerdict]: Mapping of export names to verdict objects with fields ok: bool and detail: str.

2. WebSocket & On Behalf Of (OBO) Session Tracking ​

set_ws_user ​

Pins the human user identity for the current WebSocket connection context.

  • Syntax:
    python
    pycaber.set_ws_user(
    *,
    subject: str,
    issuer: str,
    email: str | None = None,
    authenticator: str | None = None,
    acceptor: str | None = None,
    session: str | None = None,
    email_verified: bool = False
) -> str
  • Parameters:
    • subject (str, keyword only, required): User identity subject identifier sub from the IdP.
    • issuer (str, keyword only, required): Identity provider issuer URL or identifier iss conforming to RFC 9493.
    • email (str | None, keyword only, optional, default: None): Acting user email address. Passed to native Rust engine as acting_email for directory lookup fallback.
    • authenticator (str | None, keyword only, optional, default: None): Evidence provenance origin. Defaults to issuer.
    • acceptor (str | None, keyword only, optional, default: None): Target service host. Defaults to connection host.
    • session (str | None, keyword only, optional, default: None): Explicit session identifier override.
    • email_verified (bool, keyword only, optional, default: False): Whether the IdP verified the email claim (for example OIDC email_verified).
  • Return Value:
    • str: Unique pin_id token identifying this user pin.
  • Exceptions:
    • TypeError: Raised if issuer or subject is omitted.
    • ValueError: Raised if issuer or subject is empty string.
  • Behavior: Pins human identity to the active WebSocket connection. Sets ctx.user and outbound caber.user baggage header to canonical format iss_sub|<issuer>|<subject>. Once set, automatic resolvers and outbound feedback cannot override identity until released via clear_ws_user().

clear_ws_user ​

Releases a user identity pin previously registered via set_ws_user().

  • Syntax:
    python
    pycaber.clear_ws_user(pin_id: str) -> bool
  • Parameters:
    • pin_id (str, required): Pin token returned by set_ws_user().
  • Return Value:
    • bool: True if the pin was located and released; False if pin_id was unknown or expired.
  • Behavior: Unpins only the specific connection holding pin_id. Delegates directly to native Rust ConnectionHandle.unpin_user().

terminate_ws_user ​

Clears user pinning for the current WebSocket connection without requiring a pin identifier.

  • Syntax:
    python
    pycaber.terminate_ws_user() -> bool
  • Parameters: None.
  • Return Value:
    • bool: True if active connection user was unpinned; False if no WebSocket context was active.

mark_all_ws_responded ​

Closes out all open OBO request sessions for the current WebSocket connection.

  • Syntax:
    python
    pycaber.mark_all_ws_responded() -> int
  • Parameters: None.
  • Return Value:
    • int: Number of expired or closed session entries, or 0 if no WebSocket context is active.
  • Behavior: In registered content mode, purges unclaimed Rust side interaction buffer entries. In standard mode, expires stale request side n-mer rows.

register_request ​

Registers meaningful request content for On Behalf Of (OBO) chain tracking and user intent attribution.

  • Syntax:
    python
    pycaber.register_request(content: str, *, api: str | None = None, emit_synthetic: bool = True) -> InteractionHandle
  • Parameters:
    • content (str, required): The meaningful content string (for example the end user prompt, query text, or tool input).
    • api (str | None, keyword only, optional, default: None): ApiMap host identifier to target. Defaults to the current active WebSocket connection host.
    • emit_synthetic (bool, keyword only, optional, default: True): If True, emits a synthetic OBO entry if pending registration expires without an observed match.
  • Return Value:
    • InteractionHandle: Opaque handle object tying this registered request to its downstream response via register_response().
  • Behavior: Matches content against buffered incoming WebSocket messages using exact normalized substring matching with decoded text preference. Executes non blockingly. If the incoming message has not yet arrived, registers as pending and resolves when the message flows through the interceptor.

register_response ​

Registers response content and closes the active OBO interaction chain.

  • Syntax:
    python
    pycaber.register_response(content: str, *, interaction: InteractionHandle | None = None, emit_synthetic: bool = True) -> bool
  • Parameters:
    • content (str, required): The response content string (for example LLM output text, completion token stream, or final answer).
    • interaction (InteractionHandle | None, keyword only, optional, default: None): Handle returned by a prior register_request() call.
    • emit_synthetic (bool, keyword only, optional, default: True): If True, emits a synthetic OBO response record if pending expires without match.
  • Return Value:
    • bool: True if registration succeeded (match completed or pending registration created); False otherwise.
  • Behavior: Matches content against buffered outgoing messages. If the outgoing message has not been transmitted yet, registers as pending and resolves when the message flows through the interceptor.

3. Live Redaction & Privacy Controls ​

set_redact_enabled ​

Toggles payload redaction at runtime without restarting the application process.

  • Syntax:
    python
    pycaber.set_redact_enabled(enabled: bool) -> None
  • Parameters:
    • enabled (bool, required): True to enforce live payload redaction; False to enter observation only mode.
  • Return Value: None.
  • Exceptions:
    • RuntimeError: Raised if called before pycaber.configure() has initialized native runtime configuration.
  • Behavior: Mutates native global.redact_enabled atomic flag. Per API redactStrategy settings are preserved. Runtime toggle persists across configuration reloads unless new configuration explicitly defines GLOBAL.redactEnabled.

reset_redact_enabled ​

Re resolves the redaction state from environment variable CSI_GLOBAL_REDACT_ENABLED.

  • Syntax:
    python
    pycaber.reset_redact_enabled() -> None
  • Parameters: None.
  • Return Value: None.
  • Exceptions:
    • RuntimeError: Raised if called before pycaber.configure() has initialized runtime configuration.
  • Behavior: Re reads environment variable CSI_GLOBAL_REDACT_ENABLED, falls back to GLOBAL.redactEnabled in configuration, or defaults to True.

is_redact_enabled ​

Returns the current global redaction state.

  • Syntax:
    python
    pycaber.is_redact_enabled() -> bool
  • Parameters: None.
  • Return Value:
    • bool: True if redaction mutations are active; False if operating in observation only mode.
  • Exceptions:
    • RuntimeError: Raised if the installed native extension lacks redaction state accessors.

4. Agent Tool Interception & Framework Adapters ​

tool ​

Decorator syntax for wrapping custom Python functions exposed to AI agents.

  • Syntax:
    python
    @pycaber.tool(
    *,
    name: str = "",
    framework: str = "custom",
    data_domain: str = "",
    allowed_purposes: list[str] | None = None,
    captures_output: bool = True,
    prompt_allowed: bool | None = None,
    memory_allowed: bool | None = None,
    metadata: Mapping[str, Any] | None = None,
    output_rewriter: OutputRewriter | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]
  • Parameters:
    • name (str, keyword only, optional, default: ""): Tool display name.
    • framework (str, keyword only, optional, default: "custom"): Origin framework.
    • data_domain (str, keyword only, optional, default: ""): Data domain classification tag.
    • allowed_purposes (list[str] | None, keyword only, optional, default: None): Declared legitimate purpose list for privacy enforcement.
    • captures_output (bool, keyword only, optional, default: True): Whether tool return values should be captured and redacted.
    • prompt_allowed (bool | None, keyword only, optional, default: None): Whether output may flow into prompt context.
    • memory_allowed (bool | None, keyword only, optional, default: None): Whether output may be committed to long term agent memory.
    • metadata (Mapping[str, Any] | None, keyword only, optional, default: None): Custom metadata dictionary.
    • output_rewriter (OutputRewriter | None, keyword only, optional, default: None): In place object rewriter.
  • Return Value: Decorator wrapping the function with wrap_callable.

wrap_callable ​

Wraps a local callable as an interceptable internal agent action.

  • Syntax:
    python
    pycaber.wrap_callable(
    fn: Callable[..., Any],
    *,
    framework: str = "custom",
    event_type: str = "ToolInvoked",
    action_name: str = "",
    action_type: str = "tool",
    metadata: Mapping[str, Any] | None = None,
    user: str = "",
    session_id: str = "",
    output_rewriter: OutputRewriter | None = None
) -> Callable[..., Any]
  • Parameters:
    • fn (Callable[..., Any], required): Target callable function to wrap.
    • framework (str, keyword only, optional, default: "custom"): Framework origin tag.
    • event_type (str, keyword only, optional, default: "ToolInvoked"): Audit event category.
    • action_name (str, keyword only, optional, default: ""): Name of the action. Defaults to fn.__name__.
    • action_type (str, keyword only, optional, default: "tool"): Action classification string.
    • metadata (Mapping[str, Any] | None, keyword only, optional, default: None): Key value context claims.
    • user (str, keyword only, optional, default: ""): Pinned user identifier.
    • session_id (str, keyword only, optional, default: ""): Correlation session identifier.
    • output_rewriter (OutputRewriter | None, keyword only, optional, default: None): Optional callback enabling in place mutation of typed return objects.
  • Return Value: Wrapped callable with identical parameter signature.
  • Behavior: Intercepts invocation arguments for redaction before execution. Executes function. Intercepts return result or error for redaction and telemetry event emission.

process_agent_action ​

Processes a complete internal action through the synchronous interception pipeline.

  • Syntax:
    python
    pycaber.process_agent_action(capture: AgentActionCapture) -> tuple[Any, Any]
  • Parameters:
    • capture (AgentActionCapture, required): Action descriptor carrying input, output, and trace metadata.
  • Return Value:
    • tuple[Any, Any]: Pair containing (redacted_input, redacted_output).

process_agent_action_async ​

Asynchronous variant of process_agent_action.

  • Syntax:
    python
    async pycaber.process_agent_action_async(capture: AgentActionCapture) -> tuple[Any, Any]
  • Parameters:
    • capture (AgentActionCapture, required): Action descriptor.
  • Return Value:
    • Coroutine resolving to tuple[Any, Any].

register_state_resolver ​

Registers a custom framework state resolver for extracting application context.

  • Syntax:
    python
    pycaber.register_state_resolver(
    name: str,
    resolver: Callable[[Any], ResolvedState]
) -> None
  • Parameters:
    • name (str, required): Unique framework identifier (for example "streamlit", "langgraph", "crewai").
    • resolver (Callable[[Any], ResolvedState], required): Callback receiving handler or connection instance and returning ResolvedState.
  • Return Value: None.

unregister_state_resolver ​

Removes a previously registered framework state resolver.

  • Syntax:
    python
    pycaber.unregister_state_resolver(name: str) -> None
  • Parameters:
    • name (str, required): Framework identifier to remove.
  • Return Value: None.

5. Core Classes & Data Models ​

AgentActionCapture ​

Dataclass representing framework native action data prepared for Caber interception.

  • Syntax:
    python
    @dataclass
class AgentActionCapture:
    framework: str
    event_type: str
    action_name: str = ""
    action_type: str = ""
    input_data: Any = None
    output_data: Any = None
    error: str = ""
    status_code: int = 200
    metadata: dict[str, Any] = field(default_factory=dict)
    user: str = ""
    session_id: str = ""
    trace_id: str = ""
    span_id: str = ""
    api_host: str = ""
    api_path: str = ""
  • Field Definitions:
    • framework (str): Name of calling framework (for example "langchain", "autogen", "custom").
    • event_type (str): Event category (for example "ToolInvoked", "ChainExecuted").
    • action_name (str): Target tool or action identifier.
    • action_type (str): Category classification (for example "tool", "retriever", "prompt").
    • input_data (Any): Action arguments forming the synthetic request body.
    • output_data (Any): Action return value forming the synthetic response body.
    • error (str): Exception message if execution failed.
    • status_code (int): Synthetic HTTP response status code (default: 200, error: 500).
    • metadata (dict[str, Any]): Arbitrary key value security and context attributes.
    • user (str): Pinned user identifier string.
    • session_id (str): Correlation session identifier.
    • trace_id (str): OpenTelemetry hex trace identifier.
    • span_id (str): OpenTelemetry hex span identifier.
    • api_host (str): Synthetic API host identifier.
    • api_path (str): Synthetic API endpoint path.

AgentActionState ​

Mutable correlation state tracking an in flight synthetic request response pair.

  • Syntax:
    python
    @dataclass
class AgentActionState:
    capture: AgentActionCapture
    action_id: str
    trace_id: str
    span_id: str
    parent_span_id: str = ""
    started_at: float = field(default_factory=time.time)
    request_result: Any = None
  • Field Definitions:
    • capture (AgentActionCapture): Underlying action descriptor.
    • action_id (str): Unique action identifier.
    • trace_id (str): Correlation trace identifier.
    • span_id (str): In flight span identifier.
    • parent_span_id (str): Parent context span identifier.
    • started_at (float): Unix timestamp marking start of execution.
    • request_result (Any): Native result cached from request stage to pass to response stage.

ResolvedState ​

Standardized container for application level state resolved by framework adapters.

  • Syntax:
    python
    @dataclass
class ResolvedState:
    user: str = ""
    thread_id: str = ""
    agent_id: str = ""
    task_id: str = ""
    correlation_id: str = ""
    extra: dict[str, str] = field(default_factory=dict)
  • Field Definitions:
    • user (str): Human or service account principal.
    • thread_id (str): Conversation thread identifier.
    • agent_id (str): Framework agent instance identifier.
    • task_id (str): Scheduled task or workflow step identifier.
    • correlation_id (str): Distributed correlation identifier.
    • extra (dict[str, str]): Additional framework specific attributes.

CaberWSGIMiddleware ​

WSGI standard middleware for intercepting and redacting inbound HTTP requests in Flask and Django.

  • Syntax:
    python
    class CaberWSGIMiddleware:
    def __init__(self, app: Callable): ...
    def __call__(self, environ: dict, start_response: Callable) -> Iterable[bytes]: ...
  • Usage:
    python
    app.wsgi_app = CaberWSGIMiddleware(app.wsgi_app)

CaberASGIMiddleware ​

ASGI standard middleware for intercepting and redacting inbound HTTP and WebSocket requests in FastAPI and Starlette.

  • Syntax:
    python
    class CaberASGIMiddleware:
    def __init__(self, app: Callable): ...
    async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None: ...
  • Usage:
    python
    app.add_middleware(CaberASGIMiddleware)

CaberLangChainCallbackHandler ​

Lifecycle observer attaching Caber audit telemetry to LangChain chains, tools, and agents.

  • Syntax:
    python
    class CaberLangChainCallbackHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized: dict[str, Any], inputs: dict[str, Any], **kwargs: Any) -> None: ...
    def on_chain_end(self, outputs: dict[str, Any], **kwargs: Any) -> None: ...
    def on_chain_error(self, error: BaseException, **kwargs: Any) -> None: ...
    def on_tool_start(self, serialized: dict[str, Any], input_str: str, **kwargs: Any) -> None: ...
    def on_tool_end(self, output: str, **kwargs: Any) -> None: ...
    def on_tool_error(self, error: BaseException, **kwargs: Any) -> None: ...

6. Exceptions ​

RedactionUnavailable ​

Raised when a response payload cannot obtain a redaction verdict from caber_core and fail closed security is enforced.

  • Syntax: class RedactionUnavailable(RuntimeError): ...
  • Behavior: Propagates through user applications to prevent raw uninspected sensitive data leakage.