kyvvu 0.13.7

Python SDK for AI compliance monitoring and policy enforcement under the EU AI Act

Kyvvu SDK

Python SDK for instrumenting AI agents with compliance logging and runtime policy enforcement. The SDK translates agent actions into atomic behaviours, evaluates them against policies via the embedded kyvvu-engine, and streams an audit trail to the Kyvvu platform.

• Documentation: https://docs.kyvvu.com/sdk
• PyPI: pip install kyvvu

---

What the SDK does

1. Registers your agent with the Kyvvu API (POST /api/v1/agents) on startup. This is a real HTTP call that acquires an agent_id and evaluates agent-registration policies.
2. Evaluates every step your agent takes against loaded policies before execution. Policies are fetched from the API and cached (this is done by kyvvu-engine, which is included as a dependency).
3. Records completed steps into an in-memory per-task history so that path-dependent policies (predecessors, sequences, rate limits) work.
4. Flushes the audit trail to the Kyvvu API on task completion.

The SDK makes HTTP calls at two points:

• Agent registration — POST /api/v1/agents (once at startup).
• Policy fetch — GET /api/v1/policies (on first evaluation, then every KV_POLICY_TTL_SECONDS; handled by kyvvu-engine).
• Log flush — POST to the configured log endpoint (on end_task(); handled by kyvvu-engine, off by default).
• Incident webhook — POST to the configured incident endpoint (on policy violation; handled by kyvvu-engine, off by default).

---

Quickstart

from kyvvu import Kyvvu, StepType, Verb

kv = Kyvvu(api_key="KvKey-...", agent_key="my-bot")
kv.register_agent(name="My Bot", purpose="Customer support")

task_id = kv.start_task()

@kv.step(StepType.step_model, Verb.POST)
def chat(prompt: str) -> str:
    return llm.complete(prompt)

result = chat("Hello")
kv.end_task()

Bring-your-own identity

# If agent_id is provisioned externally (Terraform, config file, admin console):
kv = Kyvvu(api_key="KvKey-...", agent_key="my-bot", agent_id="ag_abc123")
# No register_agent() call needed — start using @kv.step immediately.

Task lifecycle (programmatic API)

task_id = kv.start_task()
try:
    result = my_agent_function()
except Exception as e:
    kv.error_task(error=e)
    raise
else:
    kv.end_task()

All three methods accept optional context=, properties=, and meta= kwargs for template matching and caller overrides.

---

Integration modes

The SDK supports two integration patterns. Both produce the same stream of Behavior objects; they differ in how agent actions are captured.

1. Decorator integration (@kv.step)

For custom Python agents. Wrap each function with @kv.step(step_type, verb). The decorator handles evaluate → execute → record automatically.

from kyvvu import Kyvvu, StepType, Verb, RiskClassification

kv = Kyvvu(
    api_key="KvKey-...",
    agent_key="gmail-assistant",
    risk_classification=RiskClassification.HIGH,
)
kv.register_agent(name="Gmail Assistant")

class GmailAgent:
    @kv.step(StepType.task_start)
    def fetch_email(self):
        return self._read_inbox()

    @kv.step(StepType.step_model, Verb.POST,
             properties={"model": {"name": "gpt-4o"}})
    def generate_reply(self, email):
        return llm.complete(email["body"])

    @kv.step(StepType.task_end)
    def finish(self):
        pass  # flushes audit log

See examples/custom-agent/gmail-agent/agent.py for a full working example.

2. LangChain / LangGraph callback handler

For LangChain-based agents. Construct a Kyvvu instance, register the agent, then pass a KyvvuLangChainHandler as a callback. LLM calls, tool invocations, and agent decisions are captured automatically.

from kyvvu import Kyvvu
from kyvvu.schemas import Environment, RiskClassification
from kyvvu.integrations.langchain import KyvvuLangChainHandler

kv = Kyvvu(
    api_key="KvKey-...",
    agent_key="finance-agent",
    environment=Environment.DEVELOPMENT,
    risk_classification=RiskClassification.LIMITED,
)
kv.register_agent(
    name="Finance Agent",
    purpose="Stock ticker lookup",
    metadata={"framework": "langchain", "tools": ["search"]},
)

handler = KyvvuLangChainHandler(kv)
result = agent.invoke(query, config={"callbacks": [handler]})

The handler is a pure adapter — it does not manage identity or registration. Same Kyvvu() + register_agent() pattern as the decorator integration.

See examples/langchain-agent/finance-agent/agent.py for a full working example.

Behaviour templates

Both integrations use YAML templates to map framework events to the v0.05 atomic behaviour vocabulary (step.model, step.resource, task.start, etc.). Built-in templates are provided:

from kyvvu.templates import BehaviorTemplate

dec = BehaviorTemplate.from_builtin("decorator")
lc  = BehaviorTemplate.from_builtin("langchain")

# Custom template from file
custom = BehaviorTemplate.from_path("/path/to/template.yaml")
kv = Kyvvu(api_key="...", agent_key="bot", template=custom)

Or set KV_TEMPLATE_LOCATION to a YAML file path.

Async support

The @kv.step decorator automatically detects async def functions and wraps them correctly:

@kv.step(StepType.step_model, Verb.POST)
async def chat(prompt: str) -> str:
    return await openai_client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": prompt}]
    )

Policy evaluation and recording are synchronous (sub-millisecond, in-process) — only the decorated function call is awaited. Error handling, blocked-step propagation, and task lifecycle all work identically to sync functions.

---

Project structure

kyvvu-sdk/
├── kyvvu/
│   ├── __init__.py              # Public API surface (__all__)
│   ├── __version__.py           # Version string (0.11.0)
│   ├── core.py                  # Kyvvu class — registration, task API, runner
│   ├── schemas.py               # Enums, Behavior, EvalContext (re-exports from engine)
│   ├── exceptions.py            # Exception hierarchy + KyvvuBlockedError
│   ├── logging.py               # setup_logging re-export from engine
│   ├── _task_context.py         # ContextVar for active task_id
│   ├── _limits.py               # Truncation constants for input/output capture
│   ├── decorators.py            # Re-export shim (logic in integrations/decorator.py)
│   ├── templates/
│   │   ├── __init__.py          # BehaviorTemplate class
│   │   ├── loader.py            # YAML loading and validation
│   │   ├── matcher.py           # First-match-wins rule evaluation
│   │   ├── merge.py             # Deep-merge utility
│   │   ├── helpers.py           # String interpolation for templates
│   │   ├── condition_evaluator.py
│   │   ├── decorator.template.yaml
│   │   └── langchain.template.yaml
│   ├── integrations/
│   │   ├── __init__.py          # FrameworkAdapter export
│   │   ├── _base.py             # FrameworkAdapter base class
│   │   ├── decorator.py         # @kv.step implementation
│   │   └── langchain.py         # KyvvuLangChainHandler
│   └── cli/
│       ├── main.py              # Typer app (kyvvu command)
│       ├── auth.py              # register, login, logout, whoami
│       ├── agents.py            # list-agents
│       ├── policies.py          # list-policies
│       ├── manifests.py         # list-manifests, assign-manifest, list-assignments, unassign-manifest
│       ├── config.py            # config management
│       ├── init_cmd.py          # kyvvu init (scaffold project)
│       ├── serve.py             # kyvvu serve (local engine server)
│       └── client.py            # HTTP client for CLI commands
├── tests/                       # 351 tests
│   ├── conftest.py
│   ├── test_decorator*.py       # Decorator integration tests
│   ├── test_programmatic_task_api.py
│   ├── test_identity_acquisition.py
│   ├── test_register_agent_side_effects.py
│   ├── templates/               # Template matching/loading tests
│   ├── integrations/
│   │   ├── langchain/           # 10 LangChain handler test files
│   │   └── test_framework_adapter_base.py
│   └── cli/                     # CLI command tests
├── pyproject.toml
├── pytest.ini
└── .env.example

---

Configuration

The Kyvvu constructor accepts explicit kwargs or reads from environment variables. Precedence: kwargs > env vars > .env in cwd > defaults.

Parameter	Env var	Default	Purpose
api_url	KV_API_URL	https://platform.kyvvu.com	Kyvvu API base URL
api_key	KV_API_KEY	—	Bearer API key (KvKey-...)
agent_key	KV_AGENT_KEY	—	Stable agent identifier for policy fetch
agent_id	—	—	Pre-provisioned agent ID (skips registration)
environment	KV_ENV	development	Deployment environment
risk_classification	—	minimal	EU AI Act risk tier
template	KV_TEMPLATE_LOCATION	built-in	Path to YAML template
timeout	—	10	HTTP timeout (seconds)

Engine-level settings (KV_POLICY_TTL_SECONDS, KV_LOG_ENDPOINT, etc.) are documented in the kyvvu-engine README.

---

CLI

The SDK includes a CLI (kyvvu command) for development and debugging:

kyvvu --version
kyvvu register              # create account + API key
kyvvu login                 # log in, get JWT
kyvvu logout                # clear session
kyvvu whoami                # show current user

kyvvu list-agents           # list registered agents
kyvvu list-policies         # list policies (--agent-id for per-agent)
kyvvu list-manifests        # list manifests from connected repos
kyvvu assign-manifest       # assign manifest to agent
kyvvu unassign-manifest     # remove assignment
kyvvu list-assignments      # list current assignments

kyvvu init my-agent         # scaffold a new agent project
kyvvu serve                 # start local policy evaluation server

Install CLI dependencies (included by default): typer, rich, httpx.

---

Development

# Install in editable mode (from monorepo root)
pip install -e ./kyvvu-engine && pip install -e "./kyvvu-sdk[dev,langchain]"

# Run all tests (351 tests)
cd kyvvu-sdk
python -m pytest tests/ -v

# Type checking
mypy --strict kyvvu/

# Linting
ruff check kyvvu/ tests/

---

Public API surface

The following symbols are the public API (kyvvu.__all__):

Core: Kyvvu, enrich, setup_logging

Exceptions: KyvvuError, KyvvuAPIError, KyvvuBlockedError, KyvvuConfigError, KyvvuKeyRevokedError, KyvvuRateLimitError, KyvvuTimeoutError, KyvvuValidationError

Vocabulary: StepType, Verb, Scope, Behavior, EvalContext, EvalResult, PolicyResult, Action

Agent registration: Environment, RiskClassification

Submodule exports (importable from submodules):

• kyvvu.templates.BehaviorTemplate, kyvvu.templates.deep_merge
• kyvvu.integrations.FrameworkAdapter
• kyvvu.integrations.langchain.KyvvuLangChainHandler

---

Releasing

1. Edit VERSIONS at the repo root.
2. Run ./scripts/bump-version.sh.
3. Merge to main.
4. Tag: git tag sdk-v0.x.y && git push origin sdk-v0.x.y.

---

Licence

The Kyvvu SDK is licensed under the Apache License 2.0. See LICENSE in this directory.

Important: The SDK depends on kyvvu-engine at runtime, which is separately licensed under the Business Source License 1.1 (BSL 1.1). The SDK's permissive license does not extend to the engine. Bundling, vendoring, or depending on the SDK does not grant rights to the engine beyond what BSL 1.1 permits. See kyvvu-engine/LICENSE for the engine's terms.

Production use of the engine requires a Kyvvu commercial subscription or license agreement: [e-mail]

By using Kyvvu you agree to the Terms of Service, Privacy Policy, and Acceptable Use Policy.