Skip to content

kyvvu Python API

Kyvvu Python client: instrument an agent and evaluate its actions with the Kyvvu engine.

Documentation: https://docs.kyvvu.com/

kyvvu.Action

Decision emitted by the engine after evaluating all applicable policies.

Values

allow: No violations; the intended step may proceed. warn: One or more non-critical violations; caller may proceed with caution. block: One or more critical violations; the intended step should be prevented.

Attributes:

kyvvu.Action.allow

allow = 'allow'

kyvvu.Action.block

block = 'block'

kyvvu.Action.warn

warn = 'warn'

kyvvu.ActionKind

The 12 atomic behavior types accepted by the engine.

Two namespaces. The four task.* types mark task boundaries; the eight step.* types describe agent behavior within a task. The split names a real asymmetry rather than a routing difference: a boundary marker carries no verb and neither input nor output, while a behavioral step carries the full {properties, input, output} triplet, which is what makes an absent output readable as an intended action.

Both are emitted on one path and evaluated at step_execution. TASK_BOUNDARY_KINDS names the four for callers that need to tell them apart — prefer it over matching on the task. prefix.

Attributes:

kyvvu.ActionKind.step_credential

step_credential = 'step.credential'

kyvvu.ActionKind.step_exec

step_exec = 'step.exec'

kyvvu.ActionKind.step_gate

step_gate = 'step.gate'

kyvvu.ActionKind.step_message

step_message = 'step.message'

kyvvu.ActionKind.step_model

step_model = 'step.model'

kyvvu.ActionKind.step_resource

step_resource = 'step.resource'

kyvvu.ActionKind.step_self

step_self = 'step.self'

kyvvu.ActionKind.step_unknown

step_unknown = 'step.unknown'

kyvvu.ActionKind.task_end

task_end = 'task.end'

kyvvu.ActionKind.task_error

task_error = 'task.error'

kyvvu.ActionKind.task_idle

task_idle = 'task.idle'

kyvvu.ActionKind.task_start

task_start = 'task.start'

kyvvu.Behavior

A single agent step, used for both intended and completed steps.

Intended steps (pre-execution) have output=None and seq=None. Completed steps (post-execution) have output populated and a seq assigned when the step is recorded.

input, output, properties and meta accept only I-JSON (RFC 7493) values, recursively: dict with str keys, list, str, bool, None, int with magnitude at most 2**53 - 1, and finite float. Anything else -- NaN, a larger integer, a tuple, set, bytes, datetime or Decimal -- is refused with a ValueError naming its path (input.items[2]: float NaN is not I-JSON ...), because it could not be exported, hashed or audited without being altered.

Identifiers are refused with a ValueError naming the field and the rule (task_id 'run/42': use letters, digits and . _ : - only (1-100 characters)), because from 1.1 they are hashed, signed and used in lock and storage keys.

Fields

agent_id: Registered agent identifier. Letters, digits and . _ : -, 0-255 characters (empty for an offline, unregistered runner). task_id: Current task identifier. The engine uses this to key its internal history tracker. Letters, digits and . _ : -, 1-100 characters. A task.start may carry meta["parent_task_id"], which follows the same rule. action_id: Correlates a step outcome with its evaluation. Optional on intent and may be absent on task boundary markers. Unique per (agent_id, task_id): the one legal reuse is pairing an intended step with its own completion. A duplicate is refused with ValueError. Letters, digits and . _ : -, 1-255 characters. aag_version: The AAG vocabulary version, on task.start only. Letters, digits and . + -, 1-64 characters. timestamp: When the agent initiated this step, recorded by the client. Defaults to the UTC clock at construction, so build an intended step when the intent is formed. It is also the time time-based rules (such as working_hours_only) evaluate against: a preflight is judged at the time the intent was formed, not when the action later executes. KyvvuRunner.evaluate() refuses one further than max_clock_skew_seconds from the host clock; PolicyEngine and KyvvuRunner.record() do not, so recorded history replays as is. seq: Orders a task's recorded steps: strictly increasing per (agent_id, task_id), assigned by the recording client. Consumers must not assume it is contiguous, starts at a particular value, or reflects wall-clock order; a value supplied by the caller may be ignored. None for an intended step. kind: The atomic behavior category. Set by the template. verb: Direction/mutation intent. None for step types that carry no verb (step.exec, step.gate, step.unknown, and the four task.* boundary types). step_name: Human-readable label for this step (e.g. "chat_gpt-4o"). Any I-JSON text of 1-255 characters with no control characters (U+0000-U+001F, U+007F-U+009F). input: Input data for this step. Visible to rule functions. output: Output data. Populated for completed steps; None for intended steps. A completed step.gate carries decision (pass or deny). properties: Nested key-value object carrying contextual information. Standard groups: target, auth, data, model, exec, idle, self, message, transport, raw. A step.gate carries bare kind (human or guardrail). Custom groups are allowed and passed through unchanged. meta: Correlation and provenance fields outside the behavior itself. Keys starting with kyvvu. are reserved for what the engine and the Python client write (kyvvu.eval, kyvvu.manifests); the runner refuses a caller-supplied kyvvu.* meta key in evaluate() and record().

Functions:

Attributes:

kyvvu.Behavior.aag_version

aag_version: str | None = None

kyvvu.Behavior.action_id

action_id: str | None = None

kyvvu.Behavior.agent_id

agent_id: str

kyvvu.Behavior.input

input: dict[str, Any] | None = None

kyvvu.Behavior.kind

kind: ActionKind

kyvvu.Behavior.meta

meta: dict[str, Any] | None = None

kyvvu.Behavior.model_config

model_config = ConfigDict(extra='forbid')

kyvvu.Behavior.output

output: dict[str, Any] | None = None

kyvvu.Behavior.properties

properties: dict[str, Any] = Field(default_factory=dict)

kyvvu.Behavior.seq

seq: int | None = None

kyvvu.Behavior.step_name

step_name: str

kyvvu.Behavior.task_id

task_id: str

kyvvu.Behavior.timestamp

timestamp: AwareDatetime = Field(default_factory=lambda: datetime.now(timezone.utc))

kyvvu.Behavior.verb

verb: Verb | None = None

kyvvu.Behavior.with_checked_payloads

with_checked_payloads() -> Behavior

Re-check the payload fields of a behavior built without validation.

model_copy(update=...) does not run validators, so a payload set that way (the documented intended.model_copy(update={"output": ...}) pattern, or an engine-injected meta entry) has not been checked. Recording calls this so that nothing non-I-JSON reaches the audit trail by that route.

Returns:

  • Behavior – A copy whose payloads are plain I-JSON data.

The timestamp is re-checked too: a naive datetime set that way would otherwise be exported in the host's local time.

Raises:

  • ValueError – If a payload value is not I-JSON, an identifier is outside its grammar, or the timestamp has no UTC offset.

kyvvu.BehaviorTemplate

BehaviorTemplate(data: dict[str, Any]) -> None

A loaded template ready to match events to behavior fields.

Instantiate via from_path (or the load_template alias).

Attributes:

  • name (str) – Template identifier (e.g. "langchain").
  • version (str) – Version string from the YAML file.

Functions:

  • from_path – Load a template from a YAML file.
  • match – Return behavior fields for the first matching rule.

Construct from a parsed YAML dict.

Prefer from_path over calling this directly.

Parameters:

  • data (dict[str, Any]) – Parsed template YAML dict with name, version, and rules keys.

Raises:

  • ValueError – If data is not a dict, is missing rules, or rules is not a list.
  • TemplateError – (a ValueError and a KyvvuConfigError) when a rule is malformed. See validate_rules.

kyvvu.BehaviorTemplate.from_path

from_path(path: str | Path) -> BehaviorTemplate

Load a template from a YAML file.

Parameters:

  • path (str | Path) – Filesystem path to a .yaml template file.

Returns:

Raises:

  • FileNotFoundError – When path does not exist.
  • ImportError – When pyyaml is not installed.
  • ValueError – When the file is malformed.

kyvvu.BehaviorTemplate.match

match(context: dict[str, Any]) -> dict[str, Any] | None

Return behavior fields for the first matching rule.

Evaluation stops at the first rule whose match block fully matches the context (first-match-wins). String values in the returned dict are already interpolated from context.

Parameters:

  • context (dict[str, Any]) – Event or call context dict.

Returns:

  • dict[str, Any] | None – Dict with behavior fields (kind, verb, step_name,
  • dict[str, Any] | None – properties, ...), or None when no rule matches.

kyvvu.BehaviorTemplate.name

name: str = name

kyvvu.BehaviorTemplate.rules

rules: list[dict[str, Any]]

Read-only copy of the rule dicts (without interpolation applied).

kyvvu.BehaviorTemplate.version

version: str = str(data.get('version', ''))

kyvvu.Environment

Deployment environment.

A (str, Enum) so members compare equal to their string value (Environment.PRODUCTION == "production") and serialise as the bare string via json.dumps. All values are lowercase to match the API.

Attributes:

  • DEVELOPMENT – Development / testing environment.
  • QA – Quality assurance environment.
  • STAGING – Pre-production staging environment.
  • PRODUCTION – Live production environment.

Functions:

  • all – Return all valid environment value strings.
  • is_valid – Return True if the environment string is a known value.

kyvvu.Environment.DEVELOPMENT

DEVELOPMENT = 'development'

kyvvu.Environment.PRODUCTION

PRODUCTION = 'production'

kyvvu.Environment.QA

QA = 'qa'

kyvvu.Environment.STAGING

STAGING = 'staging'

kyvvu.Environment.all

all() -> List[str]

Return all valid environment value strings.

Returns:

  • List[str] – List of valid environment values (plain str).

kyvvu.Environment.is_valid

is_valid(environment: str) -> bool

Return True if the environment string is a known value.

Accepts either a plain string or an Environment member (members compare equal to their value).

Parameters:

  • environment (str) – Value to check.

Returns:

  • bool – True if known, False otherwise.

kyvvu.EvalContext

Agent identity and execution context passed to every evaluate() call.

All fields are pre-fetched by the caller (API layer or Python client). The engine performs no I/O to retrieve these values.

Fields

agent_id: Stable string identifier for the agent (not the DB integer PK). task_id: Canonical identifier for a single agent task run. Groups all the steps recorded for one logical task. environment: Deployment environment: "development", "staging", or "production". risk_classification: EU AI Act risk tier: "high", "limited", "minimal", "unacceptable", or "unclassified". Used by classification-aware rules. user_id: Integer ID of the authenticated user who owns the agent's API key. agent_name, agent_purpose, agent_owner, agent_allowed_tools: Agent metadata pre-fetched by the API layer; forwarded to rule functions via RuleContext. manifests: Provenance records for the manifests whose policies are in effect for this agent. None when no manifest-sourced policies apply. project_root: Absolute path of the project/session working directory, when known. Supplied by integrations that run inside a project tree (e.g. kyvvu-claude populates it from the session cwd). Consumed by allowlist-containment rules such as path_within_root to block any write target that does not resolve under this root. None when the integration has no notion of a project root.

There is deliberately no evaluation-time field. Time-based rules evaluate against the behaviour's own timestamp, recorded by the client when the intent was formed, so the time a policy was checked against is the time the trace carries. The former now field is refused like any other unknown key.

Attributes:

kyvvu.EvalContext.agent_allowed_tools

agent_allowed_tools: list[str] | None = None

kyvvu.EvalContext.agent_id

agent_id: str

kyvvu.EvalContext.agent_name

agent_name: str | None = None

kyvvu.EvalContext.agent_owner

agent_owner: str | None = None

kyvvu.EvalContext.agent_purpose

agent_purpose: str | None = None

kyvvu.EvalContext.environment

environment: str = 'development'

kyvvu.EvalContext.manifests

manifests: list[ManifestRef] | None = None

kyvvu.EvalContext.model_config

model_config = ConfigDict(extra='forbid')

kyvvu.EvalContext.project_root

project_root: str | None = None

kyvvu.EvalContext.risk_classification

risk_classification: str | None = None

kyvvu.EvalContext.task_id

task_id: str | None = None

kyvvu.EvalContext.user_id

user_id: int | None = None

kyvvu.EvalResult

Aggregate result returned by PolicyEngine.evaluate().

Fields

risk_score: Normalised violation severity in [0.0, 1.0]. 0.0 means no violations; 1.0 means maximum criticality. action: Recommended action derived from risk_score: allow | warn | block. policies: One PolicyResult per evaluated policy, in evaluation order. action_id: The action_id a KyvvuRunner evaluation used or generated for a step.* behaviour; None for a task boundary and for a pure PolicyEngine evaluation. recorded_seq: Set when the engine itself recorded the evaluated step: a blocked step.* behaviour is recorded by KyvvuRunner.evaluate() and PolicyEngine.evaluate_and_record() (decision D1), and this is the seq it was given. None when nothing was recorded. A caller must not record a step whose result carries a recorded_seq.

Consumers must ignore fields they do not know: later 1.x releases may add optional fields to this model.

Attributes:

kyvvu.EvalResult.action

action: Action = Action.allow

kyvvu.EvalResult.action_id

action_id: str | None = None

kyvvu.EvalResult.policies

policies: list[PolicyResult] = Field(default_factory=list)

kyvvu.EvalResult.recorded_seq

recorded_seq: int | None = None

kyvvu.EvalResult.risk_score

risk_score: float = 0.0

kyvvu.FrameworkAdapter

FrameworkAdapter(kyvvu_instance: Kyvvu, template: object = default_template) -> None

Base class for framework integrations.

Subclasses set template_name to the name of their YAML template (e.g. "langchain") and template_package to the importable package containing it.

Parameters:

  • kyvvu_instance (Kyvvu) – An already-constructed Kyvvu instance. The adapter does not own identity acquisition -- the caller is responsible for configuring and (optionally) registering the Kyvvu instance before constructing the adapter.
  • template (object) – A pre-loaded BehaviorTemplate, which replaces the packaged template for this integration, or the default default_template (the resource named by template_name in template_package, or the file named by KV_TEMPLATE_LOCATION if that is set). None is refused with KyvvuConfigError: a callback integration cannot build behaviours without a template. The one template governs both this adapter's steps and the task boundaries it opens. It is validated at construction against boundary_contexts.

Example subclass::

class MyFrameworkHandler(FrameworkAdapter):
    template_name = "myframework"

    def on_event(self, event_data):
        ctx = {"event": "on_event", "name": event_data.name}
        self.emit_task_start(ctx)

Functions:

  • build_properties – Build the step properties dict from the matched template fields.
  • close_after_block – Close task_id as an error after a block the framework cannot see.
  • emit_step – Evaluate a pre-execution step and store the pending Behavior.
  • emit_task_end – Emit a task.end Behavior and flush the task.
  • emit_task_error – Emit a task.error Behavior and flush the task.
  • emit_task_start – Emit a task.start Behavior and return the task_id.
  • end_task_once – Call emit_fn at most once per task_id.
  • enter_tool – Expose the tool that has just started to @kv.step in this context.
  • exit_tool – Remove the tool with run_id (its end or error callback).
  • forget_root – Release every callback mapping owned by a finished root task.
  • match – Match an event context to behavior fields via the template.
  • on_block – Handle a blocked step after its recording, leaving its task open.
  • record_step – Complete a step previously started by emit_step.
  • resolve_root – Resolve the root (top-level) task_id for a nested callback.
  • safe_end_task – End a task without propagating benign flush failures to the framework.
  • task_is_open – True while task_id is a task this adapter opened and is still active.
  • tool_is_declared – Compare a callback name with client declarations.

Attributes:

kyvvu.FrameworkAdapter.boundary_contexts

boundary_contexts: Mapping[ActionKind, Sequence[Dict[str, Any]]] = {kind: ({},) for kind in COVERED_BOUNDARY_KINDS}

kyvvu.FrameworkAdapter.build_properties

build_properties(matched: Dict[str, Any], context: Dict[str, Any], step_run_id: str) -> Dict[str, Any]

Build the step properties dict from the matched template fields.

Hook for subclasses to inject framework-specific metadata into the properties before policy evaluation (e.g. LangGraph node name, CrewAI agent/task metadata). The base implementation returns the template-matched properties unchanged (as a fresh dict).

Parameters:

  • matched (Dict[str, Any]) – The template-matched behavior field dict.
  • context (Dict[str, Any]) – The original event context for this step.
  • step_run_id (str) – The pairing run identifier for this step.

Returns:

  • Dict[str, Any] – Properties dict to attach to the emitted Behavior.

kyvvu.FrameworkAdapter.close_after_block

close_after_block(task_id: str, exc: KyvvuBlockedError) -> None

Close task_id as an error after a block the framework cannot see.

A block normally leaves the task open and the framework's own error callback closes it. That fails for a block raised from a callback that is not inside a running chain: on_chain_start (the run never starts, so no error callback follows) and the root's on_chain_end (the run has already finished). Nothing else would close the task, so it would sit open until the sweeper evicted it. Records task.error without re-evaluating the policy that has just decided.

Parameters:

  • task_id (str) – The task the blocked step belongs to.
  • exc (KyvvuBlockedError) – The block, recorded as the cause of the task error.

kyvvu.FrameworkAdapter.emit_step

emit_step(context: Dict[str, Any], input_data: Optional[Dict[str, Any]], step_run_id: str, *, evaluate: bool = True) -> None

Evaluate a pre-execution step and store the pending Behavior.

The pending Behavior is keyed by step_run_id and completed later by record_step. Framework-specific property injection is delegated to build_properties; circuit-breaker behaviour on block is delegated to on_block. This single implementation is shared by every adapter — subclasses override only the hooks.

Parameters:

  • context (Dict[str, Any]) – Event context for template matching.
  • input_data (Optional[Dict[str, Any]]) – Captured input key-value pairs.
  • step_run_id (str) – Run identifier used to pair with record_step; it becomes the step's action_id. Must be unique within the task: reuse it only for the completion of this same step. A duplicate is refused: the engine raises ValueError, which propagates and records nothing.
  • evaluate (bool) – Whether to run policy evaluation. False is for an event path that can only record, such as direct CrewAI kickoff.

An event the template has no rule for is recorded as step.unknown (named after the event's name), not dropped. This is the same outcome as the catch-all rule the shipped templates end with, so a custom template behaves like them. The decorator differs: an unmatched decorated call is built from the decorator's own arguments.

A blocked step is recorded by the engine, not here.

Raises:

  • KyvvuBlockedError – If the policy engine blocks this step.
  • KyvvuConfigError – If context["task_id"] is not a task open in this client.
  • ValueError – If the engine refuses the step itself (an action_id already used in the task, a reserved kyvvu. meta key). Nothing is recorded for it.
  • TaskStateError – If the engine's runner does not have the task open (TaskClosedError when it has ended). Nothing is recorded for it.

kyvvu.FrameworkAdapter.emit_task_end

emit_task_end(context: Dict[str, Any], task_id: Optional[str] = None) -> None

Emit a task.end Behavior and flush the task.

Delegates to end_task with the matched context.

Parameters:

  • context (Dict[str, Any]) – Event context for template matching.
  • task_id (Optional[str]) – Explicit task_id to close. When None, resolved from the active task on the Kyvvu instance.

kyvvu.FrameworkAdapter.emit_task_error

emit_task_error(context: Dict[str, Any], error: Exception, task_id: Optional[str] = None) -> None

Emit a task.error Behavior and flush the task.

Delegates to error_task with the matched context.

Parameters:

  • context (Dict[str, Any]) – Event context for template matching.
  • error (Exception) – The exception that caused the task failure.
  • task_id (Optional[str]) – Explicit task_id to close.

kyvvu.FrameworkAdapter.emit_task_start

emit_task_start(context: Dict[str, Any], task_id: Optional[str] = None, *, properties: Optional[Dict[str, Any]] = None) -> str

Emit a task.start Behavior and return the task_id.

Delegates to start_task with the matched context, and remembers that this adapter owns the task.

Parameters:

  • context (Dict[str, Any]) – Event context for template matching.
  • task_id (Optional[str]) – Explicit task_id to use (e.g. LangChain run_id). When None, start_task generates a UUID. A framework-supplied id outside the task_id grammar is mapped by compliant_task_id, never refused, and the raw value is kept as properties.raw.task_id; use the returned id from then on.
  • properties (Optional[Dict[str, Any]]) – Properties merged on top of the template's for the task.start (e.g. framework correlation identifiers).

Returns:

  • str – The task_id assigned to this task.

Raises:

  • KyvvuBlockedError – If the task is blocked by policy.
  • KyvvuConfigError – If a task is already active in this context (the adapter manages its own task), or the template cannot build the boundary.

kyvvu.FrameworkAdapter.end_task_once

end_task_once(task_id: str, emit_fn: Callable[[], None]) -> None

Call emit_fn at most once per task_id.

Guards against double-flush when multiple framework callbacks try to close the same task.

Parameters:

  • task_id (str) – Task identifier to guard.
  • emit_fn (Callable[[], None]) – Zero-argument callable that emits the end/error Behavior and flushes the task.

kyvvu.FrameworkAdapter.ended_tasks

ended_tasks: set[str] = set()

kyvvu.FrameworkAdapter.enter_tool

enter_tool(name: str, run_id: str) -> None

Expose the tool that has just started to @kv.step in this context.

kyvvu.FrameworkAdapter.exit_tool

exit_tool(run_id: str) -> None

Remove the tool with run_id (its end or error callback).

Removes by run id, so nested tool callbacks unwind correctly and a callback for a tool that was never entered is a no-op.

kyvvu.FrameworkAdapter.forget_root

forget_root(root: str) -> None

Release every callback mapping owned by a finished root task.

kyvvu.FrameworkAdapter.kv

kv: Kyvvu = kyvvu_instance

kyvvu.FrameworkAdapter.match

match(context: Dict[str, Any]) -> Optional[Dict[str, Any]]

Match an event context to behavior fields via the template.

Parameters:

  • context (Dict[str, Any]) – Framework-specific event context dict.

Returns:

  • Optional[Dict[str, Any]] – Behavior field dict, or None when no rule matches. The caller
  • Optional[Dict[str, Any]] – decides what a miss means: emit_step records it as
  • Optional[Dict[str, Any]] – step.unknown rather than dropping it.

kyvvu.FrameworkAdapter.on_block

on_block(task_id: str, exc: KyvvuBlockedError) -> None

Handle a blocked step after its recording, leaving its task open.

Subclasses (e.g. CrewAI) override to trip a circuit breaker. The base implementation is a no-op.

Parameters:

  • task_id (str) – The task the blocked step belongs to.
  • exc (KyvvuBlockedError) – The block exception.

kyvvu.FrameworkAdapter.owned_tasks

owned_tasks: set[str] = set()

kyvvu.FrameworkAdapter.pending_steps

pending_steps: Dict[str, Behavior] = {}

kyvvu.FrameworkAdapter.record_step

record_step(step_run_id: str, output: Optional[Dict[str, Any]] = None) -> None

Complete a step previously started by emit_step.

Parameters:

  • step_run_id (str) – Run identifier used when emit_step was called.
  • output (Optional[Dict[str, Any]]) – Step output dict (status, result or error). If None, a default {"status": "success"} is used.

kyvvu.FrameworkAdapter.resolve_root

resolve_root(run_id: str, parent_run_id: Optional[str], *, opening_root: bool = False) -> str

Resolve the root (top-level) task_id for a nested callback.

Frameworks like LangChain nest callbacks several layers deep. The mapping is retained until the root task closes, including for nested runs whose own end callback has already arrived. Unknown or closed lineage raises before a step can be evaluated.

Parameters:

  • run_id (str) – The current callback's own run identifier.
  • parent_run_id (Optional[str]) – Direct parent run identifier, or None for top-level callbacks.
  • opening_root (bool) – True only for the callback opening a top-level task.

Returns:

  • str – String task_id of the root task for this run.

Raises:

kyvvu.FrameworkAdapter.root_run_ids

root_run_ids: Dict[str, str] = {}

kyvvu.FrameworkAdapter.runs_by_root

runs_by_root: Dict[str, set[str]] = {}

kyvvu.FrameworkAdapter.safe_end_task

safe_end_task(task_id: str, emit_fn: Callable[[], None]) -> None

End a task without propagating benign flush failures to the framework.

Like end_task_once with handling for operational flush failures.

A task-lifecycle flush (task.end / task.error) that fails for an operational reason (e.g. a transient network error in the exporter) must not propagate out of a framework callback — with raise_error=True that would crash the host chain. A KyvvuBlockedError is intentionally re-raised so that an actual policy block on the task-completion behavior still surfaces.

Parameters:

  • task_id (str) – Task identifier to guard.
  • emit_fn (Callable[[], None]) – Zero-argument callable that emits the end/error Behavior.

kyvvu.FrameworkAdapter.start_times

start_times: Dict[str, float] = {}

kyvvu.FrameworkAdapter.task_is_open

task_is_open(task_id: str) -> bool

True while task_id is a task this adapter opened and is still active.

kyvvu.FrameworkAdapter.template

template: BehaviorTemplate = _resolve_adapter_template(template, self.template_name, self.template_package, kyvvu_instance._settings.template_location)

kyvvu.FrameworkAdapter.template_name

template_name: str = 'decorator'

kyvvu.FrameworkAdapter.template_package

template_package: str = 'kyvvu.templates'

kyvvu.FrameworkAdapter.tool_is_declared

tool_is_declared(name: str, declarations: frozenset[str]) -> bool

Compare a callback name with client declarations.

kyvvu.FrameworkAdapter.warned_unmatched

warned_unmatched: set[str] = set()

kyvvu.GateDenied

Raised by a decorated step.gate function to deny.

A gate decorated with @kv.step(kind="step.gate") passes when it returns normally and records output.decision = "pass". Raising GateDenied or another exception records "deny" and propagates. A function that only prints or returns "denied" records a pass.

Deliberately not a KyvvuBlockedError (nor a KyvvuError): a block is a policy verdict on a step, a denial is the gate function's own decision, and except KyvvuBlockedError around a governed call must not swallow it.

Example::

@kv.step(kind="step.gate", properties={"kind": "human"})
def human_approval(action: str) -> None:
    if not ask_human(action):
        raise GateDenied(f"{action} refused")

kyvvu.Kyvvu

Kyvvu(api_url: Optional[str] = None, api_key: Optional[str] = None, agent_key: Optional[str] = None, agent_id: Optional[str] = None, environment: Optional[str] = None, risk_classification: str = RiskClassification.MINIMAL, *, agent_version: Optional[str] = None, framework: Optional[str] = None, declared_tools: Optional[List[str]] = None, read_only_tools: Optional[Sequence[str]] = None, write_tools: Optional[Sequence[str]] = None, policies: Optional[List[Dict[str, Any]]] = None, log_location: Optional[str] = None, log_format: Optional[str] = None, template: Union[BehaviorTemplate, _DefaultTemplate, None] = default_template, settings: Optional[KyvvuSettings] = None, timeout: int = 10) -> None

Runtime compliance client for AI agents.

Wraps KyvvuRunner with agent registration and task lifecycle management.

There are two identity-acquisition paths for agent_id:

  1. Constructor path: pass agent_id directly when you already have one (e.g. from Terraform, a config file, or the admin console).
  2. Registration path: omit agent_id and call register_agent to have the platform mint one.

These paths are mutually exclusive on a single instance.

The bundled baseline blocks tool calls unless their names are declared. Pass declared_tools to the constructor (or set KV_DECLARED_TOOLS) to declare tools without registering. A declaration passed to register_agent takes precedence when registration is used.

Parameters:

  • api_url (Optional[str]) – Base URL of the Kyvvu API (e.g. "http://localhost:8000"). Can also be set via the KV_API_URL environment variable.
  • api_key (Optional[str]) – Bearer API key (KvKey-...). Can also be set via KV_API_KEY.
  • agent_key (Optional[str]) – Agent identifier used by the runner to fetch policies.
  • agent_id (Optional[str]) – Pre-provisioned agent identifier. When supplied, no register_agent call is needed. Mutually exclusive with register_agent.
  • declared_tools (Optional[List[str]]) – Tool names this agent may invoke, including offline. Defaults to KV_DECLARED_TOOLS (comma-separated).
  • read_only_tools (Optional[Sequence[str]]) – Tool names whose callback events are reads.
  • write_tools (Optional[Sequence[str]]) – Tool names explicitly classified as writes.
  • policies (Optional[List[Dict[str, Any]]]) – Inline policy set to enforce, including offline.
  • environment (Optional[str]) – Deployment environment string. Use Environment constants. Precedence: this argument, then KV_ENVIRONMENT, then the default "production" (the same default the engine's KyvvuSettings has). Set KV_ENVIRONMENT=development for local work.
  • risk_classification (str) – EU AI Act risk tier. Use RiskClassification constants. Defaults to "minimal".
  • agent_version (Optional[str]) – Optional version of this agent, recorded on task starts.
  • framework (Optional[str]) – Optional framework name, recorded on task starts.
  • log_location (Optional[str]) – WHERE the behavioural trace goes -- a URL, a file path, "stdout", or "none". Omit it and the default "auto" follows the policy source: a client configured to fetch its policies from the platform sends its steps there too, so they appear on the dashboard's Logs page. An offline run resolves to "stdout" instead.
  • log_format (Optional[str]) – HOW the trace is formatted: "kv" (default), "json" or "otlp".
  • template (Union[BehaviorTemplate, _DefaultTemplate, None]) – Template for the decorator integration and for start_task and friends. A BehaviorTemplate replaces the built-in; the default default_template means KV_TEMPLATE_LOCATION if set, else the built-in; None means no template (@kv.step and add_step still work, built from their arguments; start_task refuses). A template must cover all four task.* types, checked at construction. Framework integrations use their own template, not this one, for the boundaries they open.
  • settings (Optional[KyvvuSettings]) – Pre-built KyvvuSettings. When supplied, individual keyword arguments are ignored.
  • timeout (int) – HTTP request timeout in seconds (default 10).

Example::

kv = Kyvvu(api_url="http://...", api_key="KvKey-...", agent_key="my-bot")
kv.register_agent(name="My Bot")

task_id = kv.start_task()
try:
    result = my_agent_function()
finally:
    kv.end_task()

Functions:

  • add_step – Record an ad-hoc compliance step without wrapping a function.
  • begin_task – Alias for start_task (backward compatibility).
  • capture_payload – Coerce a captured input / output dict to I-JSON; None stays None.
  • capture_step_name – Normalise a framework-supplied name into a valid step_name; never raises.
  • capture_value – Coerce a captured framework value to I-JSON; never raises.
  • clear_active_tools – Clear tool callback state when a framework opens a new task.
  • compliant_action_id – Map a framework-supplied id to a valid action_id; never raises.
  • compliant_task_id – Map a framework-supplied id to a valid task_id; never raises.
  • end_task – End a task and flush its remaining buffered steps to the Kyvvu API.
  • enter_tool – Mark a framework tool as running in this context.
  • error_task – End a task with an error and flush its remaining buffered steps.
  • exit_tool – Remove a completed framework tool from this context.
  • flush – Wait for queued incident and trace reports to be attempted.
  • flush_task_error – Close a task as errored without re-evaluating, exactly once.
  • idle_task – Mark a task as paused, e.g. while awaiting input.
  • make_eval_context – Build an EvalContext from this instance's settings.
  • merge_properties – Merge declared callback properties over template defaults.
  • policy_status – Return what this client is currently enforcing, and why.
  • pop_task_context – Restore the task selected before a callback.
  • push_task_context – Select an open task for a callback and return its restore token.
  • register_agent – Register this agent with the Kyvvu API.
  • report_status – Return the outbound report queue counters, or None before first use.
  • require_open_task – Raise unless task_id is a task this client has open.
  • resume_task – Set the active task for the current request context.
  • set_declared_tools – Set tool declarations found by a framework before registration.
  • start_task – Start a new compliance task.

Attributes:

  • active_task_ids (frozenset[str]) – Identifiers of tasks this client still has open.
  • agent_id (Optional[str]) –
  • current_task_id (Optional[str]) – The task_id active in the current context, or None.
  • declared_tools (Optional[List[str]]) – Tool names sent in the agent registration and evaluation context.
  • max_collection_items (int) – Maximum number of callback items captured in an input list.
  • policy_load_state (PolicyLoadState) – The PolicyLoadState of the last load.
  • read_only_tools (frozenset[str]) – Tools declared as reads for framework callback classification.
  • registration_result (Optional[EvalResult]) – The outcome of the last register_agent policy evaluation.
  • runner (KyvvuRunner) – Lazy-initialised KyvvuRunner.
  • step (Callable..., [Callable..., [Any]]) – Decorator factory for instrumenting agent steps.
  • write_tools (frozenset[str]) – Tools declared as writes for framework callback classification.

Configure a client and its policy, logging and behavior sources.

kyvvu.Kyvvu.active_task_ids

active_task_ids: frozenset[str]

Identifiers of tasks this client still has open.

kyvvu.Kyvvu.add_step

add_step(kind: Optional[Union[str, ActionKind]] = None, step_name: Optional[str] = None, *, input: Optional[Dict[str, Any]] = None, output: Optional[Dict[str, Any]] = None, verb: Optional[Union[str, Verb]] = None, properties: Optional[Dict[str, Any]] = None, meta: Optional[Dict[str, Any]] = None, task_id: Optional[str] = None, action_id: Optional[str] = None, **kwargs: Any) -> None

Record an ad-hoc compliance step without wrapping a function.

Use this when the @kv.step decorator is not practical — for example when instrumenting third-party code, dynamically chosen tools, or inline sequences where a dedicated function would be boilerplate. The caller supplies all fields explicitly; no template matching is performed.

Parameters:

  • kind (Optional[Union[str, ActionKind]]) – Step type enum value or its string representation (e.g. ActionKind.step_model or "step.model").
  • step_name (Optional[str]) – Human-readable name for this step: 1-255 characters, no control characters. Defaults to the kind value string if omitted.
  • input (Optional[Dict[str, Any]]) – Input data dict to record. Coerced to I-JSON as described in to_ijson_capture (a datetime becomes an ISO string, a tuple a list, NaN the string "NaN", ...).
  • output (Optional[Dict[str, Any]]) – Output data dict to record, coerced the same way.
  • verb (Optional[Union[str, Verb]]) – HTTP-style verb (e.g. Verb.POST or "POST").
  • properties (Optional[Dict[str, Any]]) – Arbitrary key-value metadata for the step.
  • meta (Optional[Dict[str, Any]]) – Engine-reserved multi-agent correlation fields.
  • task_id (Optional[str]) – Explicit task identifier (letters, digits and . _ : -, 1-100 characters). When omitted, the active task from start_task is used. The task must be open in this client: a step into a task that has ended, or that another process started, is refused.
  • action_id (Optional[str]) – Optional caller identifier for this action (letters, digits and . _ : -, 1-255 characters). Generated by the runner when omitted. Unique per (agent_id, task_id), except to pair an intended step with its own completion; a duplicate is refused with ValueError.
  • **kwargs (Any) – Additional fields validated by the step builder.

A task boundary kind is accepted, built and handled exactly as the matching method builds and handles it: task.start opens the task in this client (without selecting it in this context, so pass task_id= afterwards), and task.end / task.error settle open intents, record the boundary last, and close and flush the task. A boundary carries no input, output, verb or action_id. Prefer start_task, end_task and error_task.

Raises:

  • KyvvuConfigError – If agent_id is not set, no active task can be resolved, the task is not open in this client, or a task.start names a task that is already open or has ended.
  • KyvvuValidationError – If kind or verb is an unknown string value (lists the valid values), or properties / meta hold a value that is not I-JSON (names its path; these are not coerced, because policies read them), or a task boundary is given input, output, verb or action_id, or task_id, action_id or step_name is outside its grammar (see Behavior; they are refused, not mapped).
  • ValueError – If action_id is already used in the task, or meta carries a reserved kyvvu. key.
  • KyvvuBlockedError – If the step is blocked by a policy rule.

kyvvu.Kyvvu.agent_id

agent_id: Optional[str] = agent_id

kyvvu.Kyvvu.begin_task

begin_task(task_id: Optional[str] = None) -> str

Alias for start_task (backward compatibility).

Parameters:

  • task_id (Optional[str]) – Optional task identifier.

Returns:

  • str – The active task_id.

kyvvu.Kyvvu.capture_payload

capture_payload(value: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]

Coerce a captured input / output dict to I-JSON; None stays None.

kyvvu.Kyvvu.capture_step_name

capture_step_name(value: Any, fallback: str = 'unnamed') -> str

Normalise a framework-supplied name into a valid step_name; never raises.

Control characters become U+FFFD, a name over 255 characters is cut to 254 plus "…", and an empty one becomes fallback.

kyvvu.Kyvvu.capture_value

capture_value(value: Any) -> Any

Coerce a captured framework value to I-JSON; never raises.

The conversions are listed in to_ijson_capture. For framework integrations: what a callback hands over is not the caller's to fix, and recording it must not fail the host's call.

kyvvu.Kyvvu.clear_active_tools

clear_active_tools() -> None

Clear tool callback state when a framework opens a new task.

kyvvu.Kyvvu.compliant_action_id

compliant_action_id(raw: Any) -> str

Map a framework-supplied id to a valid action_id; never raises.

kyvvu.Kyvvu.compliant_task_id

compliant_task_id(raw: Any) -> str

Map a framework-supplied id to a valid task_id; never raises.

A valid id is returned unchanged; anything else becomes "x-" plus 32 hex digits of its SHA-256. Deterministic, so the same raw id always maps to the same task_id.

kyvvu.Kyvvu.current_task_id

current_task_id: Optional[str]

The task_id active in the current context, or None.

kyvvu.Kyvvu.declared_tools

declared_tools: Optional[List[str]]

Tool names sent in the agent registration and evaluation context.

kyvvu.Kyvvu.end_task

end_task(task_id: Optional[str] = None, *, context: Optional[Dict[str, Any]] = None, properties: Optional[Dict[str, Any]] = None, meta: Optional[Dict[str, Any]] = None, _template: Optional[BehaviorTemplate] = None) -> None

End a task and flush its remaining buffered steps to the Kyvvu API.

Steps are delivered at the latest by end_task(); a task may be delivered in several batches, so some may have been sent before it.

Parameters:

  • task_id (Optional[str]) – Task identifier. Resolved via current_task_id if omitted.
  • context (Optional[Dict[str, Any]]) – Event context dict for template matching.
  • properties (Optional[Dict[str, Any]]) – Caller overrides merged on top of template defaults.
  • meta (Optional[Dict[str, Any]]) – Engine-reserved multi-agent correlation fields.
  • _template (Optional[BehaviorTemplate]) – Internal. A framework integration passes its own template so that it, not this client's decorator template, builds the task boundaries the integration opens.

Raises:

  • KyvvuConfigError – If no task is active and no task_id supplied, or the task is not open in this client (already ended, or started elsewhere).

kyvvu.Kyvvu.enter_tool

enter_tool(name: str, run_id: str) -> None

Mark a framework tool as running in this context.

kyvvu.Kyvvu.error_task

error_task(task_id: Optional[str] = None, error: Optional[Exception] = None, *, context: Optional[Dict[str, Any]] = None, properties: Optional[Dict[str, Any]] = None, meta: Optional[Dict[str, Any]] = None, _template: Optional[BehaviorTemplate] = None) -> None

End a task with an error and flush its remaining buffered steps.

As with end_task, steps are delivered at the latest here; a task may be delivered in several batches.

Parameters:

  • task_id (Optional[str]) – Task identifier. Resolved via current_task_id if omitted.
  • error (Optional[Exception]) – Optional exception to record in the Behavior properties.
  • context (Optional[Dict[str, Any]]) – Event context dict for template matching.
  • properties (Optional[Dict[str, Any]]) – Caller overrides merged on top of template defaults.
  • meta (Optional[Dict[str, Any]]) – Engine-reserved multi-agent correlation fields.
  • _template (Optional[BehaviorTemplate]) – Internal. A framework integration passes its own template so that it, not this client's decorator template, builds the task boundaries the integration opens.

Raises:

  • KyvvuConfigError – If no task is active and no task_id supplied, or the task is not open in this client (already closed, or started elsewhere).

kyvvu.Kyvvu.exit_tool

exit_tool(run_id: str) -> None

Remove a completed framework tool from this context.

kyvvu.Kyvvu.flush

flush(timeout: Optional[float] = None) -> bool

Wait for queued incident and trace reports to be attempted.

end_task() and policy violations only enqueue their reports; a background worker sends them, so neither waits on the network. Call this before a short-lived script exits if you need the trace to have left the process. It is also run automatically at interpreter exit, bounded by KV_REPORT_EXIT_FLUSH_SECONDS (default 2). A hard kill (SIGKILL) can lose reports still in the queue.

Does not start a runner: a client that never governed a step has nothing queued and returns True at once.

Parameters:

  • timeout (Optional[float]) – Seconds to wait. None waits until the queue is empty.

Returns:

  • bool – True when every report queued so far has been attempted,
  • bool – False when the deadline passed first. True does not mean
  • bool – the platform acknowledged anything: a send that failed was still
  • bool – attempted, and is logged rather than reported here.

kyvvu.Kyvvu.flush_task_error

flush_task_error(task_id: str, exc: BaseException) -> None

Close a task as errored without re-evaluating, exactly once.

Records a task.error Behavior directly and ends the task. Unlike error_task, this does not run the policy engine on the error Behavior.

A policy block does not call this. The engine records a blocked step itself and raises, leaving the task open, so that path history (taint, gates, sequences) survives and the caller can approve and retry. This remains for a caller that has decided to abandon a task without a policy pass.

The task.error is built by the same template path as error_task, so both close a task in one shape. A client built with template=None has no boundary rules; it records a plain task.error instead.

Idempotent: a task that is not active is left alone, so a second call records nothing and flushes nothing.

Parameters:

  • task_id (str) – The task to flush. No-op if falsy or not active.
  • exc (BaseException) – The exception that caused the flush.

kyvvu.Kyvvu.idle_task

idle_task(task_id: Optional[str] = None, *, context: Optional[Dict[str, Any]] = None, properties: Optional[Dict[str, Any]] = None, meta: Optional[Dict[str, Any]] = None, _template: Optional[BehaviorTemplate] = None) -> None

Mark a task as paused, e.g. while awaiting input.

The task stays open: memory is assumed to persist across the idle period, so this records a task.idle marker without ending the task, discarding it from _active_tasks, or clearing the active-task context. There is no explicit resume marker — the next action on the task signals resumption (AAG docs/model.md §2.1).

Parameters:

  • task_id (Optional[str]) – Task identifier. Resolved via current_task_id if omitted.
  • context (Optional[Dict[str, Any]]) – Event context dict for template matching.
  • properties (Optional[Dict[str, Any]]) – Caller overrides merged on top of template defaults.
  • meta (Optional[Dict[str, Any]]) – Engine-reserved multi-agent correlation fields.
  • _template (Optional[BehaviorTemplate]) – Internal. A framework integration passes its own template so that it, not this client's decorator template, builds the task boundaries the integration opens.

Raises:

  • KyvvuConfigError – If no task is active and no task_id supplied, or the task is not open in this client.
  • KyvvuBlockedError – If a policy blocks the idle marker.

kyvvu.Kyvvu.make_eval_context

make_eval_context(agent_id: str, task_id: Optional[str] = None) -> EvalContext

Build an EvalContext from this instance's settings.

Collapses the otherwise-identical EvalContext construction used by register_agent / start_task / end_task / error_task / add_step.

Parameters:

  • agent_id (str) – Agent identifier for the context.
  • task_id (Optional[str]) – Task identifier, or None for evaluation at the agent_registration enforcement point (which has no task).

Returns:

kyvvu.Kyvvu.max_collection_items

max_collection_items: int

Maximum number of callback items captured in an input list.

kyvvu.Kyvvu.merge_properties

merge_properties(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]

Merge declared callback properties over template defaults.

kyvvu.Kyvvu.policy_load_state

policy_load_state: PolicyLoadState

The PolicyLoadState of the last load.

Shorthand for kv.policy_status().load_state. The three states that mean nothing is being enforced are none_configured and unavailable; enforcing and stale_cache both mean at least one policy is in effect. See policy_status for the check to write.

kyvvu.Kyvvu.policy_status

policy_status() -> PolicyStatus

Return what this client is currently enforcing, and why.

A policy count cannot tell an agent with nothing assigned apart from an agent whose policy fetch failed, and both allow every step. Read load_state for that distinction. A policy set that contains a policy which cannot be loaded is not a third case: it raises PolicyLoadError out of the load, so there is no partly loaded set to inspect.

The state reflects the last load. Policies are fetched lazily on the first governed step, so call this after register_agent (or after a step has run) rather than straight after construction, or the answer will be unavailable simply because nothing has been loaded yet.

Returns:

Example::

from kyvvu import PolicyLoadState

GOVERNED = (PolicyLoadState.enforcing, PolicyLoadState.stale_cache)
if kv.policy_load_state not in GOVERNED:
    raise SystemExit("refusing to run ungoverned")

stale_cache belongs in that set: it means policies are being enforced but the platform could not be reached to confirm they are current. Treating it as ungoverned would take an agent down over a transient API blip. Only none_configured and unavailable mean nothing is being enforced.

registration_blocked is a separate question: whether the agent's registration is blocked by a policy. The runner holds what register_agent evaluated and evaluates it again whenever a policy refresh changes the agent's agent_registration policies, so a manifest assigned to a running agent applies without registering again. While blocked, every step raises KyvvuBlockedError naming the policies in registration_blocked_by; a later evaluation that passes clears it.

kyvvu.Kyvvu.pop_task_context

pop_task_context(token: Token[Optional[str]]) -> None

Restore the task selected before a callback.

kyvvu.Kyvvu.push_task_context

push_task_context(task_id: str) -> Token[Optional[str]]

Select an open task for a callback and return its restore token.

kyvvu.Kyvvu.read_only_tools

read_only_tools: frozenset[str]

Tools declared as reads for framework callback classification.

kyvvu.Kyvvu.register_agent

register_agent(name: Optional[str] = None, purpose: Optional[str] = None, owner_id: Optional[str] = None, declared_tools: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, description: Optional[str] = None, store_metadata: bool = False) -> Dict[str, Any]

Register this agent with the Kyvvu API.

Posts to POST /api/v1/agents. Sets agent_id from the platform response, then evaluates policies whose enforcement point is agent_registration in-process via the engine, using the current registration metadata even when the platform does not store it.

Caching: Successful registrations are cached on disk at ~/.kyvvu/registrations/<agent_key>.json. On subsequent startups, if the registration payload has not changed, the cached agent_id is reused without a network call. Set the KV_REGISTRATION_TTL environment variable (e.g. "24h", "7d") to force periodic re-registration.

Retry: Transient failures (5xx, timeouts, connection errors) are retried up to 3 times with exponential backoff (1 s, 3 s, 9 s). Uses time.sleep() — blocks the calling thread.

Stale fallback: If all retries fail and a stale cache exists, the cached agent_id is used with a warning. The cached response is evaluated with the current metadata against the available agent_registration policies; a blocking policy still prevents registration. With no cache at all (a first run, or an ephemeral HOME) there is nothing to fall back on and the error is raised: registration fails closed, and its message names the URL tried and the offline alternatives.

Mutually exclusive with passing agent_id to the constructor.

agent_key, environment, and risk_classification are taken from the constructor -- they are not parameters here.

Parameters:

  • name (Optional[str]) – Human-readable display name.
  • purpose (Optional[str]) – Description of what the agent does.
  • owner_id (Optional[str]) – Owner contact string.
  • declared_tools (Optional[List[str]]) – List of tool/step names this agent may invoke.
  • metadata (Optional[Dict[str, Any]]) – Extra key-value pairs stored with the agent. Must be I-JSON; registration policies read it, so it is refused rather than converted.
  • description (Optional[str]) – Free-text description of the agent.
  • store_metadata (bool) – When True, name/purpose/owner_id/maintainer_id/ metadata are stored in Kyvvu's cloud. When False (default), only the opaque agent ID and operational fields are stored server-side.

Returns:

  • Dict[str, Any] – API response dict (includes "agent_id", "agent_key", etc.).

Raises:

  • KyvvuConfigError – If agent_id was passed to constructor, or if api_url / api_key is not configured.
  • KyvvuValidationError – If metadata is not I-JSON.
  • KyvvuAPIError – On HTTP errors (after retries exhausted, with no stale cache available).
  • KyvvuKeyRevokedError – On HTTP 401 (immediate, no retry).
  • KyvvuRateLimitError – On HTTP 429 (immediate, no retry).
  • KyvvuTimeoutError – On connection timeout (after retries exhausted, with no stale cache available).
  • KyvvuBlockedError – If an agent_registration policy blocks the response, including when using a stale cache.

kyvvu.Kyvvu.registration_result

registration_result: Optional[EvalResult]

The outcome of the last register_agent policy evaluation.

register_agent evaluates the agent_registration policies and raises KyvvuBlockedError on a block, but a warn outcome only reaches warnings.warn and the incident sink, and its return value is the platform's response, not the evaluation. This is where a caller reads which policies warned -- to show them in its own interface, say -- without parsing a warning message.

Set on every path that evaluates: a live registration, a cached one and the stale-cache fallback. A block sets it to the blocking result (the same object as the exception's result) before raising. A registration that fails before evaluating -- a refused request, an unreachable API with no cache -- leaves it None, never the previous call's outcome.

Returns:

  • The (Optional[EvalResult]) – class:~kyvvu_engine.schemas.EvalResult of the last
  • Optional[EvalResult] – registration evaluation, or None when the last
  • Optional[EvalResult] – register_agent call did not reach one (or none was made).

kyvvu.Kyvvu.report_status

report_status() -> Optional[ReportStatus]

Return the outbound report queue counters, or None before first use.

dropped counts reports refused because the queue (KV_REPORT_QUEUE_SIZE, default 1000) was full.

kyvvu.Kyvvu.require_open_task

require_open_task(task_id: str) -> None

Raise unless task_id is a task this client has open.

Every write into a task -- a step, task.idle, task.end, task.error -- goes through this check. A task is open between start_task and end_task / error_task on this client instance, in this process. Writing into anything else would start a second, independently numbered record of the task: after it ended, or from another process that never opened it. Neither is supported in 1.0; how a task may be continued elsewhere is left to a later version to define, which is why it is refused rather than tolerated.

This refuses; it never guesses. It does not pick another open task, and the caller still names the task (by context or task_id=).

Raises:

kyvvu.Kyvvu.resume_task

resume_task(task_id: str) -> None

Set the active task for the current request context.

Call at the start of an HTTP handler to associate it with a task that was started in a previous request. The task must have been started on this client, in this process, and not yet ended: this selects an open task for the current context, it does not take over a task another client or process opened. 1.0 has no way to continue a task elsewhere.

Parameters:

  • task_id (str) – A task_id from a prior start_task() call on this client.

Raises:

kyvvu.Kyvvu.runner

runner: KyvvuRunner

Lazy-initialised KyvvuRunner.

Returns:

  • KyvvuRunner – The shared KyvvuRunner instance.

kyvvu.Kyvvu.set_declared_tools

set_declared_tools(tools: Sequence[str]) -> None

Set tool declarations found by a framework before registration.

kyvvu.Kyvvu.start_task

start_task(task_id: Optional[str] = None, *, context: Optional[Dict[str, Any]] = None, properties: Optional[Dict[str, Any]] = None, meta: Optional[Dict[str, Any]] = None, _template: Optional[BehaviorTemplate] = None, _framework: Optional[str] = None) -> str

Start a new compliance task.

Multiple concurrent tasks are allowed on the same instance (e.g. one per user session in a web application).

Parameters:

  • task_id (Optional[str]) – Optional task identifier: letters, digits and . _ : - only, 1-100 characters. Auto-generated (UUID4) if omitted. A task_id must never be reused: one that is open in this client is refused, and one that has ended, or was used by another process, must not be passed again -- the platform would receive two unrelated records under one identifier.
  • context (Optional[Dict[str, Any]]) – Event context dict for template matching.
  • properties (Optional[Dict[str, Any]]) – Caller overrides merged on top of template defaults.
  • meta (Optional[Dict[str, Any]]) – Engine-reserved multi-agent correlation fields.
  • _template (Optional[BehaviorTemplate]) – Internal. A framework integration passes its own template so that it, not this client's decorator template, builds the task boundaries the integration opens.

Returns:

  • str – The active task_id.

Raises:

  • KyvvuConfigError – If agent_id is missing, task_id is already open in this client or was already ended by it, or this client was built with template=None and no integration template is passed.
  • KyvvuValidationError – If task_id is not letters, digits and . _ : - only, 1-100 characters. It is refused, not mapped.
  • KyvvuBlockedError – If the task.start step is blocked by policy (both ContextVar and instance state are rolled back).

kyvvu.Kyvvu.step

step: Callable[..., Callable[..., Any]]

Decorator factory for instrumenting agent steps.

Returns:

  • Callable..., [Callable..., [Any]] – step(kind, verb=None, ...) callable.

Example::

@kv.step(ActionKind.step_resource, Verb.GET)
def fetch_data(url):
    return requests.get(url).json()

kyvvu.Kyvvu.write_tools

write_tools: frozenset[str]

Tools declared as writes for framework callback classification.

kyvvu.KyvvuAPIError

KyvvuAPIError(message: str, status_code: Optional[int] = None, response: Optional[dict[str, object]] = None) -> None

Raised when API request fails.

This exception is raised for HTTP errors, connection failures, and other API-related issues.

Attributes:

  • message – Error message describing what went wrong.
  • status_code – HTTP status code if applicable (None for connection errors).
  • response – API response body if available (None for connection errors).

Example::

try:
    kv.register_agent(name="my-agent")
except KyvvuAPIError as e:
    print(f"API error: {e}")
    if e.status_code:
        print(f"HTTP status: {e.status_code}")

Initialize API error.

Parameters:

  • message (str) – Error message.
  • status_code (Optional[int]) – HTTP status code if applicable.
  • response (Optional[dict[str, object]]) – API response body if available.

kyvvu.KyvvuAPIError.response

response = response

kyvvu.KyvvuAPIError.status_code

status_code = status_code

kyvvu.KyvvuBlockedError

KyvvuBlockedError(result: EvalResult, step_name: str = '') -> None

Raised by KyvvuRunner.evaluate() when the engine returns Action.block.

For a step.* behaviour the runner has already recorded the blocked step when this is raised (decision D1): result.recorded_seq is its seq and result.action_id its action_id. The caller must not record it again; a second record() of that action_id is refused as a duplicate. recorded_seq is None for a blocked task boundary, which is not recorded.

Attributes:

Retain the denying evaluation and format its user-facing step message.

kyvvu.KyvvuBlockedError.result

result = result

kyvvu.KyvvuBlockedError.step_name

step_name = step_name

kyvvu.KyvvuClientUpgradeRequired

KyvvuClientUpgradeRequired(message: str, info: ClientUpgradeInfo) -> None

The platform refused this client as too old to serve.

Raised when the policy fetch (or registration) answers 426 with code: client_upgrade_required. A PolicyLoadError, because it is one: the platform's current set cannot be loaded by this client. That puts it under the single policy-load tolerance point -- a runner that already holds a validated set (in memory, or in its on-disk cache) keeps enforcing it, reported as stale_cache with stale_cause=client_upgrade_required; only a cold start with nothing to fall back on reaches the caller.

The message is the one line a person needs, never a list of param errors.

Attributes:

  • info – What the platform said, as a ClientUpgradeInfo.

Attach the platform's compatibility requirements to the refusal.

kyvvu.KyvvuClientUpgradeRequired.info

info = info

kyvvu.KyvvuConfigError

Raised when the Python client is misconfigured.

This exception is raised for configuration issues such as: - Missing API key - Missing agent_id when a step is invoked - Conflicting identity-acquisition paths

Example::

try:
    kv = Kyvvu(api_key=None)
except KyvvuConfigError as e:
    print(f"Configuration error: {e}")

kyvvu.KyvvuError

Base exception for all Kyvvu Python client errors.

All Python client exceptions inherit from this class, allowing you to catch all Kyvvu-related errors with a single except clause.

Example::

try:
    kv.register_agent(name="my-agent")
except KyvvuError as e:
    print(f"Kyvvu error occurred: {e}")

kyvvu.KyvvuKeyRevokedError

KyvvuKeyRevokedError(message: str, response: Optional[dict[str, object]] = None) -> None

Raised when the API key has been disabled or revoked.

This exception is ALWAYS raised regardless of settings -- agents must halt immediately when their key is revoked or deleted.

Attributes:

  • message – Human-readable message including the reason from the API.
  • status_code – Always 401.
  • response – API response body if available.

Example::

try:
    kv.register_agent(name="my-agent")
except KyvvuKeyRevokedError as e:
    print(f"Agent halted -- API key revoked: {e}")

Initialize key revoked error.

Parameters:

  • message (str) – Error message describing the revocation.
  • response (Optional[dict[str, object]]) – API response body if available.

kyvvu.KyvvuRateLimitError

KyvvuRateLimitError(message: str, retry_after: Optional[int] = None) -> None

Raised when rate limit is exceeded.

This exception is raised when the API returns a 429 status code indicating that the rate limit has been exceeded.

Attributes:

  • message – Error message.
  • retry_after – Number of seconds to wait before retrying (if provided by API).

Example::

import time

try:
    kv.register_agent(name="my-agent")
except KyvvuRateLimitError as e:
    if e.retry_after:
        time.sleep(e.retry_after)

Initialize rate limit error.

Parameters:

  • message (str) – Error message.
  • retry_after (Optional[int]) – Seconds to wait before retrying.

kyvvu.KyvvuRateLimitError.retry_after

retry_after = retry_after

kyvvu.KyvvuTimeoutError

Raised when API request times out.

This exception is raised when a request to the API exceeds the configured timeout value.

Example::

try:
    kv = Kyvvu(api_key="KvKey-...", timeout=5)
    kv.register_agent(name="my-agent")
except KyvvuTimeoutError as e:
    print(f"Request timed out: {e}")

kyvvu.KyvvuValidationError

Raised when data validation fails.

This exception is raised when payload validation fails, such as invalid enum values or missing required fields.

Example::

from kyvvu.schemas import AgentRegistrationPayload

try:
    payload = AgentRegistrationPayload(
        agent_key="my-agent",
        risk_classification="INVALID",
    )
    payload.validate()
except KyvvuValidationError as e:
    print(f"Validation error: {e}")

kyvvu.PolicyLoadError

PolicyLoadError(message: str, errors: list[str] | None = None) -> None

A policy, or a manifest of policies, was refused at load time.

Raised for a policy whose params do not match its rule's params_schema (a required key missing, a key the rule does not know, a value of the wrong type), and for the retired scope key. It is raised on every load path - PolicyEngine.load_policies, manifest parsing, the API manifest validator's assignment check and the Python client's policy_file - and never degraded to a cached or partial policy set: a configuration that loads fewer policies than it names enforces less than it claims.

Subclasses ValueError so a caller that already guarded manifest loading with except ValueError keeps working.

Attributes:

  • errors – One message per problem found, each naming the policy, the rule and the offending key. Loading collects every problem before raising, so an author fixes the file once rather than once per run.

Keep the overall failure message and its individual validation errors.

kyvvu.PolicyLoadError.errors

errors = list(errors or [])

kyvvu.PolicyLoadState

Why the policy set currently in the engine looks the way it does.

A policy count cannot distinguish situations that demand very different responses: an agent with nothing assigned, an agent whose policy fetch failed, and an agent whose entire policy set was rejected as unloadable all report zero policies and all evaluate as allow. This enum is the missing distinction.

Values

enforcing: At least one policy is loaded and being enforced. none_configured: The load succeeded and the source genuinely has no policies. Nothing is being enforced, and that is the source's answer, not a failure. unavailable: Nothing is loaded. Either no load has been attempted yet, or the fetch failed and no cached set was available to fall back on. stale_cache: Policies are being enforced, but the set is not known to match its source: the last fetch failed (whether the set in effect came from memory or from a disk cache), or the disk cache that is the configured source has outlived policy_cache_max_age_seconds. Staleness is a property of the load, not of which source the set came from — the long-lived agent whose API went down keeps enforcing what it has, and this is the state that says so.

There is no "some policies were dropped" state: a policy that cannot be loaded raises PolicyLoadError out of the load, so a set is either loaded whole or refused whole.

Attributes:

kyvvu.PolicyLoadState.enforcing

enforcing = 'enforcing'

kyvvu.PolicyLoadState.none_configured

none_configured = 'none_configured'

kyvvu.PolicyLoadState.stale_cache

stale_cache = 'stale_cache'

kyvvu.PolicyLoadState.unavailable

unavailable = 'unavailable'

kyvvu.PolicyResult

Per-policy evaluation outcome within a single EvalResult.

Fields

policy_id: ID of the evaluated policy. name: Human-readable policy name. severity: Policy severity level. violated: True if the rule function returned False or raised an exception. violation_details: Rule-specific details when violated; None when the policy passed. manifest_id: Assignment row id of the manifest this policy came from; None for hand-authored policies. manifest_name: Human-readable source manifest name, when available.

Attributes:

kyvvu.PolicyResult.manifest_id

manifest_id: int | None = None

kyvvu.PolicyResult.manifest_name

manifest_name: str | None = None

kyvvu.PolicyResult.name

name: str

kyvvu.PolicyResult.policy_id

policy_id: int | None

kyvvu.PolicyResult.rule_type

rule_type: str = ''

kyvvu.PolicyResult.severity

severity: str

kyvvu.PolicyResult.violated

violated: bool

kyvvu.PolicyResult.violation_details

violation_details: dict[str, Any] | None = None

kyvvu.PolicySourceKind

Where a configuration is able to obtain policies from.

This is a question about configuration, not about runtime state: it says which source a caller could in principle reach, not whether that source answered. Resolved by policy_source_kind and reported as source once a load has happened.

Values

api: The Kyvvu platform API (api_url + api_key + agent_key). local: Policies the caller supplied itself, with no credentials and no network: an inline list (policies) or one or more manifest YAML files on disk (policy_file). This is the credential-free source — it is what pip install to a visible block runs on, and it is resolved before disk_cache because a caller who names a policy file is naming a source, whereas a cache path names a place a fetch may once have written to. disk_cache: An on-disk policy cache (policy_cache_path), written by an earlier successful API fetch. none: No policy source at all — such a configuration can never enforce anything, which is why KyvvuRunner refuses to be constructed from it.

Attributes:

kyvvu.PolicySourceKind.api

api = 'api'

kyvvu.PolicySourceKind.disk_cache

disk_cache = 'disk_cache'

kyvvu.PolicySourceKind.local

local = 'local'

kyvvu.PolicySourceKind.none

none = 'none'

kyvvu.PolicyStatus

Status of the runner's policy cache.

Returned by policy_status.

Fields

loaded: True if policies have been loaded at least once. last_success: Wall-clock time of last successful policy fetch. last_attempt: Wall-clock time of last fetch attempt (success or failure). policy_count: Number of active policies in the engine. stale: True if the last fetch failed and the cache has exceeded TTL. source: Where the current policies came from: "api", "local", "disk_cache", or "none". last_fetch_at: Alias for last_success (backward compatibility). last_fetch_succeeded: Whether the most recent fetch attempt succeeded. instance_id: The runner's instance identifier. ttl_remaining_seconds: Seconds until the in-memory cache expires and a re-fetch is triggered. load_state: Which PolicyLoadState the last load ended in. This is the field that tells "no policies assigned" apart from "fetch failed" and from "the source answered with nothing"; policy_count alone cannot. policies_offered: How many policies were handed to the loader. Read together with policy_count; they differ only transiently, because a policy that cannot be loaded refuses the whole load. registration_blocked_by: Names of the agent_registration policies whose violation blocked the agent's registration when it was last evaluated, at register_agent() or again after a refresh changed the registration policies. Empty when it is not blocked. registration_blocked: Whether the registration is blocked. While it is, every step.* evaluation is refused with KyvvuBlockedError naming those policies; task boundaries are not. Cleared by a later evaluation of the registration that does not block. governed: Whether at least one validated policy is in effect. stale_cause: Why the set in effect is not known to match its source, while it is not; None when it does. client_upgrade_required is the one that means pip install -U. client_upgrade: What the platform said when it last refused this client as too old, or None. Cleared by the next accepted fetch. agent_id: The agent this runner governs: the platform agent_id the policy set was served for, when the platform is the source and has said; otherwise the agent_id the runner bound to on its first evaluation or record; None before either. With the platform as the source this is the agent_id a caller must send, and any other is refused with AgentBindingError. applicable_policy_count: How many of the loaded policies apply to agent_id: the workspace-wide ones plus those naming that agent, counted the way policy_count is. None while agent_id is. Equal to policy_count for the agent a set was fetched for; lower means some loaded policies never apply to this agent.

Attributes:

kyvvu.PolicyStatus.agent_id

agent_id: str | None = None

kyvvu.PolicyStatus.applicable_policy_count

applicable_policy_count: int | None = None

kyvvu.PolicyStatus.client_upgrade

client_upgrade: dict[str, Any] | None = None

kyvvu.PolicyStatus.governed

governed: bool

Whether the current load state has policies in effect.

kyvvu.PolicyStatus.instance_id

instance_id: str = ''

kyvvu.PolicyStatus.last_attempt

last_attempt: datetime | None = None

kyvvu.PolicyStatus.last_fetch_at

last_fetch_at: datetime | None = None

kyvvu.PolicyStatus.last_fetch_succeeded

last_fetch_succeeded: bool = False

kyvvu.PolicyStatus.last_success

last_success: datetime | None = None

kyvvu.PolicyStatus.load_state

load_state: PolicyLoadState = PolicyLoadState.unavailable

kyvvu.PolicyStatus.loaded

loaded: bool = False

kyvvu.PolicyStatus.policies_offered

policies_offered: int = 0

kyvvu.PolicyStatus.policy_count

policy_count: int = 0

kyvvu.PolicyStatus.registration_blocked

registration_blocked: bool

Whether the agent's registration is blocked by a policy.

kyvvu.PolicyStatus.registration_blocked_by

registration_blocked_by: list[str] = Field(default_factory=list)

kyvvu.PolicyStatus.source

source: str = 'none'

kyvvu.PolicyStatus.stale

stale: bool = False

kyvvu.PolicyStatus.stale_cause

stale_cause: StaleCause | None = None

kyvvu.PolicyStatus.ttl_remaining_seconds

ttl_remaining_seconds: float = 0.0

kyvvu.RiskClassification

EU AI Act risk tier (Article 6).

A (str, Enum) so members compare equal to their string value (RiskClassification.HIGH == "high"). All values are lowercase to match the API.

Attributes:

  • HIGH – High-risk AI systems (e.g. hiring, credit scoring).
  • LIMITED – Limited-risk systems (e.g. chatbots).
  • MINIMAL – Minimal-risk systems (e.g. spam filters).
  • UNACCEPTABLE – Prohibited systems.
  • UNCLASSIFIED – Not yet classified (default).

Functions:

  • all – Return all valid risk classification value strings.
  • is_valid – Return True if the classification string is a known value.

kyvvu.RiskClassification.HIGH

HIGH = 'high'

kyvvu.RiskClassification.LIMITED

LIMITED = 'limited'

kyvvu.RiskClassification.MINIMAL

MINIMAL = 'minimal'

kyvvu.RiskClassification.UNACCEPTABLE

UNACCEPTABLE = 'unacceptable'

kyvvu.RiskClassification.UNCLASSIFIED

UNCLASSIFIED = 'unclassified'

kyvvu.RiskClassification.all

all() -> List[str]

Return all valid risk classification value strings.

Ordered minimal, limited, high, unacceptable, unclassified to preserve the historical ordering used in validation error messages.

Returns:

  • List[str] – List of valid risk classification values (plain str).

kyvvu.RiskClassification.is_valid

is_valid(classification: str) -> bool

Return True if the classification string is a known value.

Accepts either a plain string or a RiskClassification member.

Parameters:

  • classification (str) – Value to check.

Returns:

  • bool – True if known, False otherwise.

kyvvu.StaleCause

Why the set in effect is not known to match its source.

PolicyLoadState says that a set is stale; this says why, because the remedies differ: a network problem, a policy definition this engine refuses, and an engine the platform no longer serves are three different fixes, and only the last is pip install -U.

Values

fetch_failed: The last fetch did not complete (network, timeout, an HTTP error). set_refused: The source answered with a set this engine refuses to load (PolicyLoadError). client_upgrade_required: The platform refused this client as too old to serve (KyvvuClientUpgradeRequired). The set in effect is the last one fetched before that. cache_expired: The configured disk cache has outlived policy_cache_max_age_seconds. local_source_unreadable: The configured manifest could not be read. caller_marked: A caller of load_policies passed stale=True.

Attributes:

kyvvu.StaleCause.cache_expired

cache_expired = 'cache_expired'

kyvvu.StaleCause.caller_marked

caller_marked = 'caller_marked'

kyvvu.StaleCause.client_upgrade_required

client_upgrade_required = 'client_upgrade_required'

kyvvu.StaleCause.fetch_failed

fetch_failed = 'fetch_failed'

kyvvu.StaleCause.local_source_unreadable

local_source_unreadable = 'local_source_unreadable'

kyvvu.StaleCause.set_refused

set_refused = 'set_refused'

kyvvu.TASK_BOUNDARY_KINDS

TASK_BOUNDARY_KINDS: frozenset[ActionKind] = frozenset({ActionKind.task_start, ActionKind.task_end, ActionKind.task_error, ActionKind.task_idle})

kyvvu.TaskClosedError

Raised when a write names a task that is closing or closed.

A task is closing once its task.end or task.error is recorded: only end_task(), which finishes cleanup, may follow. It is closed by end_task(), or evicted by the stale-task sweeper. Within one runner (or PolicyEngine) a closed task_id is not reopened, for its most recent 100,000 closed tasks: a second record under the same identifier would restart seq and be indistinguishable from the first. Older ids may be forgotten, and nothing is claimed across processes; a task_id must still never be reused.

kyvvu.TaskStateError

Raised when a write does not fit the task lifecycle.

A task is opened by recording its task.start. A KyvvuRunner refuses any other write (evaluate, record, restore_intent) into a task it has not opened, and refuses a second task.start for a task it has already opened. TaskClosedError is the subclass for a write into a task that is closing or closed.

kyvvu.Verb

HTTP-style verb describing the direction and mutation intent of a step.

Attributes:

kyvvu.Verb.DELETE

DELETE = 'DELETE'

kyvvu.Verb.GET

GET = 'GET'

kyvvu.Verb.PATCH

PATCH = 'PATCH'

kyvvu.Verb.POST

POST = 'POST'

kyvvu.__version__

Kyvvu Python client version information.

kyvvu.default_template

default_template: _DefaultTemplate = _DefaultTemplate()

kyvvu.get_current_task_id

get_current_task_id() -> Optional[str]

Return the task_id active in the current context, or None.

This reads the ContextVar set by start_task (or by framework adapters) and is safe to call from any thread or async task. Useful for advanced users who need to inspect the active task without a Kyvvu instance reference.

Returns:

  • Optional[str] – The active task_id, or None if no task is active in this context.

kyvvu.load_manifest_policies

load_manifest_policies(paths: str | Path | list[str | Path]) -> list[dict[str, Any]]

Load one or more manifests into policy dicts, refusing an invalid one.

The convenience form for every caller that wants policies rather than a validation report: the local policy source on KyvvuSettings, kyvvu try, and kyvvu serve --policy-file.

Parameters:

  • paths (str | Path | list[str | Path]) – A single path, a comma-separated list of paths (see PATH_LIST_SEPARATOR), or a list of either — every string element is comma-split, so the two spellings behave identically. Policies are returned in the order the manifests are given, so a later manifest never displaces an earlier one — every entry is loaded and the engine evaluates all of them.

Returns:

  • list[dict[str, Any]] – Policy dicts ready for
  • list[dict[str, Any]] – meth:~kyvvu_engine.engine.PolicyEngine.load_policies, each carrying
  • list[dict[str, Any]] – its manifest's name as manifest_name.

Raises:

  • PolicyLoadError – When a policy's params do not match its rule's schema or it carries the retired scope key. Never a ManifestError, so a caller that degrades around an unusable file does not degrade around this.
  • ManifestError – When a file is unreadable, or when a manifest is otherwise invalid. Refusing an invalid manifest rather than loading the entries that happened to parse is deliberate: a partially loaded policy set is a failure mode where a configuration looks populated and enforces less than it claims.

kyvvu.register_client

register_client(name: str, version: str, *, sdk_version: str | None = None) -> None

Name the integration this process is, for X-Kyvvu-Client.

The Python client registers kyvvu/<version> when it is imported; an integration built on it (kyvvu-langchain, kyvvu-crewai) registers itself afterwards and wins, because the integration package is the one a person would upgrade. kyvvu-claude, which uses the engine without the Python client, registers itself.

Parameters:

  • name (str) – The distribution name, such as kyvvu-claude.
  • version (str) – Its installed version.
  • sdk_version (str | None) – The Python client's version, when it is present. Left as it is when None, so an integration does not erase what the Python client set.

kyvvu.setup_logging

setup_logging(level: str | None = None, json: bool = False, force: bool = False) -> None

Configure the root logger.

Parameters:

  • level (str | None) – Log level string ("DEBUG", "INFO", "WARNING", "ERROR"). Defaults to the KV_LOG_LEVEL environment variable, or "INFO" if that is not set.
  • json (bool) – Emit JSON-formatted lines when True. Defaults to False (human-readable plaintext).
  • force (bool) – Re-configure even if setup_logging() was already called. Useful in tests that need a specific level.