SDK

The Kyvvu Python SDK instruments your agent with two things: a register_agent() call at startup, and @kv.log_step() decorators on each method you want to track. The SDK manages task IDs, step numbering, internal variable capture, and error handling — you just describe what each step is.

pip install kyvvu

---

Initialization

from kyvvu import Kyvvu

kv = Kyvvu(
    api_key="KvKey-your-api-key",       # or set KYVVU_API_KEY env var
    api_url="http://localhost:8000",     # or set KYVVU_API_URL env var
)

Both api_key and api_url can be omitted if the corresponding environment variables are set (KYVVU_API_KEY, KYVVU_API_URL). The API URL defaults to http://localhost:8000 if neither is provided.

Error and incident handling can be configured at init time:

kv = Kyvvu(
    api_key="KvKey-...",
    on_error="raise",                        # "raise" | "log" | "ignore"  (default: "raise")
    on_incident="log",                       # "raise" | "log" | "ignore"  (default: "log")
    incident_severity_threshold="critical",  # minimum severity to trigger on_incident
    timeout=10,                              # request timeout in seconds
)

on_error controls what happens when the API is unreachable or returns an error. on_incident controls what happens when a policy violation is detected. Setting on_incident="raise" halts execution and raises KyvvuPolicyViolationError; "log" records the violation and continues; "ignore" does nothing.

---

Agent registration

Call register_agent() once at startup, before any steps are logged:

kv.register_agent(
    agent_key="my-agent",                  # stable ID you choose
    name="My Agent",
    purpose="Handles customer support tickets",
    owner_id="[email protected]",
    risk_classification="limited",         # "minimal" | "limited" | "high" | "unacceptable"
    environment="production",              # "development" | "qa" | "staging" | "production" | "copilot"
    metadata={"version": "1.0.0"},
)

Registration is idempotent — calling it again with the same agent_key updates the existing record. If any agent_registration policies are violated, the behaviour depends on your on_incident setting. To catch violations explicitly:

from kyvvu.exceptions import KyvvuPolicyViolationError

try:
    kv.register_agent(agent_key="my-agent", ...)
except KyvvuPolicyViolationError as e:
    for incident in e.incidents:
        print(f"[{incident['severity']}] {incident['title']}")

You can use the RiskClassification and Environment enums from kyvvu.schemas instead of raw strings if you prefer type safety:

from kyvvu.schemas import RiskClassification, Environment

kv.register_agent(
    agent_key="my-agent",
    risk_classification=RiskClassification.HIGH,
    environment=Environment.PRODUCTION,
    ...
)

---

Logging steps with @kv.log_step

Decorate any method with @kv.log_step(node_type) to log it as a step. The decorator captures inputs, outputs, execution time, and internal variables automatically. Your function body is unchanged.

@kv.log_step("START_NODE")
def start(self, user_input: str) -> str:
    return f"Processing: {user_input}"

@kv.log_step("LLM_CALL")
def generate(self, prompt: str) -> str:
    response = self.llm.complete(prompt)
    return response

@kv.log_step("END_NODE")
def finish(self, result: str) -> None:
    print(f"Done: {result}")

Valid node types: START_NODE, END_NODE, LLM_CALL, TOOL_CALL, MEMORY_CALL, HUMAN_APPROVAL, PII_CHECK, DECISION, ERROR. Use NodeType from kyvvu.schemas for autocomplete.

Task lifecycle — the SDK generates a new task_id each time a START_NODE step runs and clears it when END_NODE runs. All steps in between are automatically numbered and grouped under that task. Always start with START_NODE and end with END_NODE. If a new START_NODE is called before the previous task's END_NODE, the open task is automatically logged as INTERNAL_TASK_UNFINISHED.

Decorator options

@kv.log_step(
    node_type="TOOL_CALL",
    node_name="fetch_account",          # defaults to function name if omitted
    is_external_call=True,              # marks this step as calling an external service
    has_write_permission=True,          # marks this step as having write access
    capture_internal=False,             # disable internal variable capture (faster)
    metadata={"api": "stripe"},         # static metadata added to every log entry
    node_attributes={"endpoint": "..."},# policy-queryable attributes
)
def fetch_account(self, customer_id: str) -> dict:
    ...

capture_internal=True (the default) uses sys.settrace to capture local variables inside the function body. This is useful for development and debugging — it gives you a full picture of what the function was doing. Disable it in performance-critical production paths with capture_internal=False.

Errors are logged automatically. If a decorated function raises an exception, the decorator logs an ERROR entry with the exception type, message, and stack trace before re-raising. You don't need to wrap steps in try/except to get error visibility.

---

Enriching steps with enrich()

enrich() lets you attach metadata to the current step from anywhere inside the call stack — useful when the data you want to log is generated deep inside a function.

from kyvvu import enrich

@kv.log_step("LLM_CALL")
def generate(self, prompt: str) -> str:
    response = self.llm.complete(prompt)
    # Attach token usage and cost to this step's log entry
    enrich({
        "model": "gpt-4o-mini",
        "prompt_tokens": response.usage.prompt_tokens,
        "completion_tokens": response.usage.completion_tokens,
        "cost_usd": 0.002,
    })
    return response.text

enrich() is a module-level function — no need to pass kv around. The data is merged into the step's meta field when the decorator logs the step. It uses Python's contextvars so it's safe in concurrent contexts.

---

Exceptions

All SDK exceptions inherit from KyvvuError, so you can catch everything with a single clause:

from kyvvu.exceptions import KyvvuError, KyvvuPolicyViolationError, KyvvuAPIError, KyvvuConfigError

try:
    kv.register_agent(...)
except KyvvuPolicyViolationError as e:
    # Policy violated — e.incidents contains the full list
    ...
except KyvvuAPIError as e:
    # HTTP error or connection failure — e.status_code available
    ...
except KyvvuConfigError as e:
    # Misconfiguration (missing API key, agent not registered, etc.)
    ...
except KyvvuError as e:
    # Catch-all for any Kyvvu error
    ...

KyvvuPolicyViolationError is only raised if on_incident="raise" and the violation meets or exceeds incident_severity_threshold. By default (on_incident="log"), violations are logged as warnings and execution continues.

---

Full example

from kyvvu import Kyvvu, enrich
from kyvvu.exceptions import KyvvuPolicyViolationError
import os

kv = Kyvvu(api_key=os.getenv("KYVVU_API_KEY"))

try:
    kv.register_agent(
        agent_key="support-agent",
        name="Support Agent",
        purpose="Classifies and responds to support tickets",
        owner_id="[email protected]",
        risk_classification="limited",
        environment="production",
    )
except KyvvuPolicyViolationError as e:
    print(f"Registration policy violation: {e}")


class SupportAgent:

    @kv.log_step("START_NODE")
    def start(self, ticket: dict) -> dict:
        return ticket

    @kv.log_step("LLM_CALL", is_external_call=True)
    def classify(self, ticket: dict) -> str:
        category = self.llm.classify(ticket)
        enrich({"model": "gpt-4o-mini", "category": category})
        return category

    @kv.log_step("TOOL_CALL", is_external_call=True, has_write_permission=False)
    def fetch_context(self, ticket: dict) -> dict:
        return self.crm.lookup(ticket["customer_id"])

    @kv.log_step("HUMAN_APPROVAL")
    def await_approval(self, draft: str) -> str:
        return draft  # surface for review in your UI

    @kv.log_step("END_NODE")
    def finish(self, outcome: str) -> None:
        pass

For a complete worked example with branching logic, see Custom Python Agent.