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.

---

Basic usage

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

from kyvvu import Kyvvu, StepType, 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")

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

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

When chat("Hello") is called:

1. The decorator constructs a Behavior with step_type=step.model, verb=POST, 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, KyvvuBlockedError is raised and the function does not execute.

Task lifecycle

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

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

@kv.step(StepType.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

Argument	Type	Default	Description
step_type	StepType	required	The atomic behaviour type (StepType.step_model, StepType.step_resource, etc.).
verb	Verb	None	HTTP-style verb (Verb.GET, Verb.POST, etc.).
step_name	str	function name	Human-readable step name. Defaults to the decorated function's __name__.
properties	dict	{}	Structured metadata merged into the Behavior's properties.
meta	dict	None	Framework-specific metadata (parent task ID, etc.).
capture_input	bool	True	Whether to capture function arguments as the Behavior's input.
capture_output	bool	True	Whether to capture the return value as the Behavior's output.

Properties example

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

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

Concurrent tasks in web servers

In web servers where multiple requests are handled concurrently, the SDK 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()

Handling blocks

When a policy blocks a step, the decorator raises KyvvuBlockedError:

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})")
    # Decide: retry with different approach, ask for approval, abort

How the decorator uses templates

The decorator passes its arguments through the template system before constructing a Behavior. The template can:

• Map step_name to additional properties (e.g. "if step_name starts with db_, set target.system = postgres").

• Override step_type based on naming conventions.

• Set default properties for all steps of a given type.

To use a custom template:

from kyvvu.templates import BehaviorTemplate

template = BehaviorTemplate.from_path("my_template.yaml")
kv = Kyvvu(api_key="...", agent_key="bot", template=template)

---

Next steps

• LangChain / LangGraph — callback-based integration for LangChain agents

• Templates — how template matching and deep-merge work

• Tasks and History — task lifecycle and path-dependent evaluation