Python Decorator¶

What you'll learn: How to use @kv.step to instrument custom Python agents, including task lifecycle, async support, and all available options.

kyvvu is the Python client: it registers the agent, runs its tasks, and evaluates each step with the engine in your process. The decorator on this page is its most direct use. The LangChain and LangGraph and CrewAI integrations build on the same client, so the tasks, options and verdicts described here apply to them too.

Basic usage¶

Wrap each function with @kv.step(kind, verb). The decorator handles the evaluate-execute-record cycle automatically.

from kyvvu import Kyvvu, ActionKind, Verb

kv = Kyvvu(api_url="https://platform.kyvvu.com", api_key="KvKey-...", agent_key="my-bot")
kv.register_agent(
    name="My Bot",
    purpose="Customer support triage and response",
    declared_tools=["read_ticket"],
)

@kv.step(ActionKind.step_model)
def chat(prompt: str) -> str:
    return llm.complete(prompt)

@kv.step(ActionKind.step_resource, Verb.GET)
def read_ticket(ticket_id: str) -> dict:
    return db.get_ticket(ticket_id)

declared_tools matters. Every registered agent gets the bundled baseline, whose allowlist policy blocks any step.resource, step.exec or step.credential whose step name is not on that list, and blocks all of them when the list is empty. A decorated step's name is the function's name, so read_ticket is declared above; chat is a step.model and needs no entry. Without the declaration, the first read_ticket(...) raises KyvvuBlockedError: ... Tool calls must be in the declared allowlist. The baseline also wants a purpose of at least 30 characters, and warns at registration when it is shorter.

Steps belong to a task. A decorated function called outside kv.start_task() / kv.end_task() raises KyvvuConfigError("no active task; call kv.resume_task(task_id) in this context") — it is not evaluated, and it does not run. end_task() is also what flushes: it queues the task's buffered steps for delivery to the platform (or your configured log_location), and without it nothing is persisted. end_task() does not wait for delivery, which a background worker does, so a short script should finish with kv.flush() (an exit flush waits about 2 s as a backstop).

task_id = kv.start_task()
try:
    chat("Hello")
finally:
    kv.end_task(task_id)

When chat("Hello") is called inside a task: 1. The decorator constructs a Behavior with kind=step.model, verb=None, step_name="chat". 2. It calls evaluate() — the engine checks policies against the task history. 3. If allowed, the function executes. 4. The completed step (with output) is recorded into history via record(). 5. If blocked, the blocked step is recorded, KyvvuBlockedError is raised and the function does not execute. The task stays open.

Task lifecycle¶

Use ActionKind.task_start and ActionKind.task_end to mark task boundaries:

@kv.step(ActionKind.task_start)
def begin_task(self):
    return self._read_inbox()

@kv.step(ActionKind.task_end)
def finish(self):
    pass  # triggers log flush

Or use the 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 lifecycle methods (start_task, end_task, error_task) accept optional context=, properties=, and meta= kwargs for template matching and caller overrides.

Decorator arguments¶

Decorator steps use the explicit verb= argument. The read_only_tools and write_tools constructor options apply to LangChain, LangGraph and CrewAI tool callbacks, not to decorated steps.

Properties example¶

Include get_customer in the declared_tools passed to register_agent() before calling this resource step, just as read_ticket is declared above.

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

@kv.step(ActionKind.step_resource, Verb.GET,
         properties={"target": {"system": "salesforce", "table": "customer-data"}})
def get_customer(self, customer_id):
    return sf.query(customer_id)

Properties are deep-merged with template output. See Templates for merge semantics.

Async support¶

The decorator automatically detects async def functions:

@kv.step(ActionKind.step_model)
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 work identically to sync functions.

Concurrent tasks in web servers¶

In web servers where multiple requests are handled concurrently, the Python client uses a ContextVar to track the active task_id per execution context. Each thread or asyncio task gets its own task.

# Each request handler starts its own task:
task_id = kv.start_task()
# ... all @kv.step calls in this context use this task_id ...
kv.end_task()

A new request or thread does not inherit the task. Call kv.resume_task(task_id) before a decorated step, for a task this client started and has not ended, or pass task_id= to kv.add_step() and the task methods. A context with no selected task never attaches to a lone open task: the step raises KyvvuConfigError.

Handling blocks¶

When a policy blocks a step, the engine records the blocked step (output.status is blocked, with meta["kyvvu.eval"]), the decorator raises KyvvuBlockedError, and leaves the task open. A block never ends a task, on the decorator or on any other surface:

from kyvvu import KyvvuBlockedError

try:
    result = run_dangerous_action()
except KyvvuBlockedError as exc:
    print(f"Blocked: {exc}")
    print(f"Risk score: {exc.result.risk_score}")
    for p in exc.result.policies:
        if p.violated:
            print(f"  - {p.name} ({p.severity})")

The caller decides what comes next:

• Give up on the task: kv.error_task(task_id) records task.error and closes it, or kv.end_task(task_id) closes it normally. error_task on a task that is already closed raises.
• Get approval and retry in the same task: record a step.gate (a decorated gate, or kv.add_step(kind="step.gate", ...)), then call the action again. The blocked attempt stays in the history but is ignored by step_requires_gate and step_requires_predecessor, and step_directly_preceded_by looks past it.
• Do nothing: a task nobody closes is evicted by the stale-task sweeper after KV_TASK_MAX_AGE_SECONDS.

Gates: pass on a normal return¶

A function decorated with kind="step.gate" passes only when it returns normally. A normal return records output.decision = "pass"; raising kyvvu.GateDenied records "deny" and propagates to the caller. A function that only prints or returns "denied" records a pass, so deny by raising:

from kyvvu import GateDenied, ActionKind

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

Any other exception also records output.decision = "deny" and propagates. GateDenied is not a KyvvuBlockedError: a block is a policy's verdict on a step, a denial is the gate function's own decision, so except KyvvuBlockedError around a governed call does not swallow it. properties.kind defaults to human; set it to guardrail for an automated check. The decorator records the decision even when capture_output=False.

Do not decorate a framework tool¶

When a framework handler (KyvvuLangChainHandler and friends) runs a tool, it already records that tool. Decorating the tool function with @kv.step would record the call twice, so calling it raises KyvvuConfigError. Decorate a helper the tool calls instead, or use kv.add_step(...) inside the tool. See LangChain / LangGraph.

How the decorator uses templates¶

The decorator passes its arguments through the template system before constructing a Behavior. The match context has kind, verb (when declared), name (the function's name) and kwargs (the function's arguments by parameter name). The template can: - Map the function's name to additional properties (e.g. "if name starts with db_, set target.system = postgres"), with name_pattern. - Override kind based on naming conventions. - Set default properties for all steps of a given type.

step_name is an output field of a rule, not something a rule can match on; to match on the function's name use name or name_pattern. {{ kwargs.url }} resolves only if the function has a parameter called url.

To use a custom template:

from kyvvu import Kyvvu
from kyvvu_templates import BehaviorTemplate

template = BehaviorTemplate.from_path("my_template.yaml")
kv = Kyvvu(api_url="https://platform.kyvvu.com", api_key="KvKey-...", agent_key="bot", template=template)
kv.register_agent(
    name="Bot",
    purpose="Customer support triage and response",
    declared_tools=["read_ticket"],
)

A custom template replaces the built-in decorator template, and it also governs start_task(), end_task(), error_task() and idle_task(), so it must match all four task.* types (by kind). Start from kyvvu template show decorator > my_template.yaml (the kyvvu command, from kyvvu-cli). The template changes how steps are built, not where the agent's id comes from: start_task() still needs register_agent() (or Kyvvu(agent_id=...)) first, and raises KyvvuConfigError: no agent_id without it. template=None means no template: @kv.step and kv.add_step still work, built from their own arguments, but start_task() refuses because there is no rule to build a boundary from. See Writing a Custom Template.

Next steps¶

• LangChain / LangGraph — callback-based integration for LangChain agents
• CrewAI — event-listener integration for CrewAI crews
• Templates — how template matching and deep-merge work
• Tasks and History — task lifecycle and path-dependent evaluation