Custom Python Agent

This example walks through a complete Kyvvu integration using the Python SDK decorator pattern. The agent is a customer support ticket router: it fetches a ticket, classifies it with an LLM, routes it down one of three specialized paths, generates a response, requests human approval, and closes the task. Seven steps, real branching logic, all logged.

The full source is in examples/custom-agent/ in the platform repo.

---

What this example covers

• Initializing the Kyvvu client and registering an agent

• Decorating methods with @kv.log_step() to log each step

• Handling branching workflows (different paths, same audit trail)

• Graceful handling of policy violations at registration time

• Running the agent end-to-end

---

Setup

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

kv = Kyvvu(
    api_key=os.getenv("KV_DEMO_API_KEY"),
    api_url=os.getenv("KV_DEMO_API_URL", "http://localhost:8000"),
)

Kyvvu is the entry point for everything. One instance per agent process is the normal pattern. It holds the API connection, tracks the current task, and provides the @kv.log_step decorator.

---

Agent registration

Agents register themselves at startup by calling kv.register_agent(). This creates (or updates) the agent record in Kyvvu and triggers any agent_registration policies immediately.

try:
    kv.register_agent(
        agent_key="customer-support-router_1",   # stable identifier for this agent
        name="Customer Support Agent",
        purpose="Classifies and routes customer support tickets",
        owner_id="[email protected]",
        risk_classification="limited",
        environment="development",
        metadata={
            "version": "1.0.0",
            "llm_provider": "openai",
        },
    )
except KyvvuPolicyViolationError as e:
    # Registration succeeded but one or more policies were violated.
    # The agent is visible in the dashboard and incidents have been created.
    # You decide whether to halt or continue.
    print(f"Policy violation at registration: {e}")
    for incident in e.incidents:
        print(f"  [{incident['severity']}] {incident['title']}")

A few things to note:

• agent_key is the stable identifier you choose. Kyvvu hashes it together with your API key to produce a unique agent_id. Re-registering with the same agent_key updates the existing record rather than creating a duplicate.

• risk_classification accepts minimal, limited, high, or unacceptable. Policies can be scoped to specific risk levels — a high-risk agent may face stricter checks than a limited-risk one.

• Catching KyvvuPolicyViolationError is optional but recommended in production. If you don't catch it, the exception propagates and your agent won't start.

---

Logging steps with @kv.log_step

Every meaningful operation in the agent is decorated with @kv.log_step(node_type). The decorator intercepts the function call, wraps it in a logged step, and forwards the return value unchanged. Your agent code does not change.

from openai import OpenAI

class CustomerSupportAgent:

    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.current_ticket = None
        self.ticket_category = None
        self.current_response = None

    # Step 1 — START_NODE
    @kv.log_step("START_NODE")
    def fetch_ticket(self) -> dict:
        """Fetch the next ticket from the queue."""
        # In production: call your ticketing system API
        self.current_ticket = {
            "ticket_id": "TICKET-1001",
            "customer_email": "[email protected]",
            "subject": "Unable to login to my account",
            "message": "I keep getting 'invalid credentials' but my password is correct.",
            "priority": "high",
        }
        return self.current_ticket

    # Step 2 — LLM_CALL
    @kv.log_step("LLM_CALL")
    def classify_ticket(self, ticket: dict) -> str:
        """Use an LLM to classify the ticket as technical, billing, or general."""
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": (
                    "Classify the ticket as exactly one of: technical, billing, general. "
                    "Reply with the category name only."
                )},
                {"role": "user", "content": f"Subject: {ticket['subject']}\n{ticket['message']}"},
            ],
            max_tokens=10,
            temperature=0.3,
        )
        category = response.choices[0].message.content.strip().lower()
        self.ticket_category = category
        return category

The decorator records the step's node_type, captures the function's inputs and outputs, timestamps the execution, and chains the log entry's hash to the previous step's hash. None of this requires any change to your function body.

---

Branching workflows

Real agents don't follow a straight line. This agent routes to one of three specialized paths based on the LLM classification — but the audit trail captures all of it regardless of which branch runs.

    # Step 3a — TOOL_CALL (technical path only)
    @kv.log_step("TOOL_CALL")
    def create_jira_ticket(self, ticket: dict) -> str:
        """Escalate a technical issue to the engineering queue."""
        jira_id = f"ENG-{random.randint(1000, 9999)}"  # mock
        self.jira_ticket_id = jira_id
        return jira_id

    # Step 3b — TOOL_CALL (billing path only)
    @kv.log_step("TOOL_CALL")
    def lookup_account(self, customer_email: str) -> dict:
        """Retrieve account data for a billing query."""
        return {
            "account_id": f"ACC-{customer_email.split('@')[0].upper()}",
            "subscription_tier": "Premium",
            "last_charge": "$29.99",
        }

    # Step 4 — LLM_CALL (all paths)
    @kv.log_step("LLM_CALL")
    def generate_response(self, ticket: dict, category: str, context: dict = None) -> str:
        """Generate a category-appropriate response draft."""
        prompts = {
            "technical": "You are a technical support specialist. Be empathetic and solution-focused.",
            "billing":   "You are a billing specialist. Be reassuring about payment concerns.",
            "general":   "You are a friendly support agent. Be warm and helpful.",
        }
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": prompts[category]},
                {"role": "user",   "content": f"Respond to: {ticket['subject']}\n{ticket['message']}"},
            ],
            max_tokens=300,
        )
        self.current_response = response.choices[0].message.content.strip()
        return self.current_response

    # Step 5 — HUMAN_APPROVAL
    @kv.log_step("HUMAN_APPROVAL")
    def get_current_response(self):
        """Surface the draft for human review before sending."""
        return self.current_response

    # Step 6 — TOOL_CALL
    @kv.log_step("TOOL_CALL")
    def send_response(self, ticket_id: str, response_text: str) -> dict:
        """Send the approved response to the customer."""
        return {"status": "sent", "ticket_id": ticket_id}

    # Step 7 — END_NODE
    @kv.log_step("END_NODE")
    def reset_state(self):
        """Close the task and clear state for the next ticket."""
        self.current_ticket = None
        self.ticket_category = None
        self.current_response = None

Each @kv.log_step call is independent. The SDK tracks which task is running and automatically assigns the correct step number — you don't manage sequencing yourself. Steps that don't run (e.g., create_jira_ticket on a billing ticket) simply aren't logged, which is exactly what you want in the audit trail.

---

Running the workflow

With the class defined, the execution loop is straightforward:

if __name__ == "__main__":
    agent = CustomerSupportAgent(api_key=os.getenv("OPENAI_API_KEY"))

    # Fetch and classify
    ticket   = agent.fetch_ticket()
    category = agent.classify_ticket(ticket)

    # Route based on category
    context = {}
    if category == "technical":
        jira_id = agent.create_jira_ticket(ticket)
        context = {"jira_id": jira_id}
    elif category == "billing":
        account = agent.lookup_account(ticket["customer_email"])
        context = {"account_data": account}

    # Generate, approve, send
    draft  = agent.generate_response(ticket, category, context)
    review = agent.get_current_response()   # HUMAN_APPROVAL step
    agent.send_response(ticket["ticket_id"], draft)
    agent.reset_state()                      # END_NODE — closes the task

After this runs, the Kyvvu dashboard will show a complete task with all logged steps in order, their inputs and outputs, and the validated hash chain proving the log hasn't been tampered with.

---

What the dashboard shows

Once the agent has run, navigate to Logs in the dashboard and find the task. You'll see:

• Each step listed in sequence with its node_type, timestamp, and duration

• Input and output data captured for each step

• Hash chain validation status — every step shows ✓ if the chain is intact

• Any incidents generated by step_execution or task_execution policies

If you have a policy configured (e.g., "every LLM_CALL must be preceded by a PII_CHECK"), violations appear as incidents linked directly to the offending step.

---

Key patterns to take away

One Kyvvu instance, module-level. Initialize kv once at the top of your agent module. All decorated methods share the same client and task context.

Register at startup, not per-request. kv.register_agent() is called once when the agent process starts. It's idempotent — safe to call on every restart.

END_NODE closes the task. The SDK tracks an open task from START_NODE until it sees END_NODE. Always end your workflow with an END_NODE-decorated method, even if it just clears state. A task without an END_NODE is flagged as unfinished.

Branching is fine. You don't need to tell Kyvvu about your routing logic. Decorate every meaningful operation and let the audit trail reflect what actually ran.

The decorator is transparent. @kv.log_step does not modify return values and adds negligible latency. What happens if a log call fails (network error, API down) is controlled by the on_error setting on the Kyvvu instance — see SDK Reference for details.

---

Next steps

• LangChain agent: If you're using LangChain, the callback handler approach is even less intrusive — see LangChain Agent

• Policies: Set up a step_execution policy to enforce that HUMAN_APPROVAL always precedes TOOL_CALL in your workflow — any task that skips the approval step will generate an incident automatically

• Audit reports: After a few runs, generate a PDF report from the dashboard under Reports → Generate