> For the complete documentation index, see [llms.txt](https://docs.kyvvu.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.kyvvu.com/integrations/decorator.md).

# 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.

```python
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)
```

**Steps belong to a task.** A decorated function called outside `kv.start_task()` / `kv.end_task()` raises `KyvvuConfigError("no active task; call kv.start_task() before decorated step functions")` — it is not evaluated, and it does not run. `end_task()` is also what **flushes**: the buffered steps reach the platform (or your configured `log_location`) there and nowhere else, so without it nothing is persisted.

```python
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 `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:

```python
@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:

```python
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

```python
@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](/core-concepts/templates.md) for merge semantics.

## Async support

The decorator automatically detects `async def` functions:

```python
@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.

```python
# 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`:

```python
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:

```python
from kyvvu_engine.templates import BehaviorTemplate

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

***

## Next steps

* [LangChain / LangGraph](/integrations/langchain.md) — callback-based integration for LangChain agents
* [Templates](/core-concepts/templates.md) — how template matching and deep-merge work
* [Tasks and History](/core-concepts/tasks.md) — task lifecycle and path-dependent evaluation