REST API Integration

If you're building an agent in a language other than Python, or you prefer not to use the SDK, you can integrate directly with the Kyvvu API using two endpoints:

• POST /api/v1/agents — register or update your agent

• POST /api/v1/logs — log a step

That's it. Everything else (policy evaluation, incident creation, hash chain validation, dashboard visibility) happens server-side automatically.

This example uses curl to show the raw requests, but the same logic applies in any language or framework.

---

Authentication

All agent requests authenticate with an API key in the Authorization header:

Authorization: Bearer KvKey-your-api-key-here

Your API key is set in .env as KV_SEED_ROOT_API_KEY on local development, or created via the dashboard under Settings → API Keys in production.

---

Step 1 — Register your agent

Call POST /api/v1/agents once at startup. Pass your agent_key — a stable string you choose that identifies this agent — along with descriptive fields.

curl -X POST http://localhost:8000/api/v1/agents \
  -H "Authorization: Bearer KvKey-your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "agent_key": "my-rest-agent",
    "name": "My REST Agent",
    "purpose": "Demonstrates direct REST API integration with Kyvvu",
    "owner_id": "[email protected]",
    "risk_classification": "limited",
    "environment": "development"
  }'

The response includes the agent_id you'll use in all subsequent log calls:

{
  "agent_id": "agent_a1b2c3d4e5f6...",
  "agent_key": "my-rest-agent",
  "registration_status": "created",
  "incidents": []
}

How agent_id is generated

The agent_id is a deterministic hash of your agent_key and your API key. This means:

• The same agent_key + API key always produces the same agent_id

• Re-registering with the same agent_key updates the existing agent rather than creating a new one — registration_status will be "updated" on subsequent calls

• Different API keys produce different agent_id values even for the same agent_key — agents are scoped to the account that owns the API key

In practice: store the agent_id from the first registration response and reuse it. Or just re-register at every startup — it's idempotent.

---

Step 2 — Log a task

A task is one end-to-end execution of your agent. You manage the task_id yourself: generate a UUID at the start of each run and use it for all steps in that run.

A step is one operation within a task. Steps must be numbered sequentially starting at 1. The first step should be START_NODE and the last should be END_NODE.

Each log entry must include a prev_hash — the hash of the previous step's log entry — to form the tamper-evident chain. The very first step uses an empty string as prev_hash. The hash of each step is returned in the response and must be passed as prev_hash in the next call.

Step 1 — START_NODE

TASK_ID=$(python3 -c "import uuid; print(str(uuid.uuid4()))")

curl -X POST http://localhost:8000/api/v1/logs \
  -H "Authorization: Bearer KvKey-your-api-key" \
  -H "Content-Type: application/json" \
  -d "{
    \"agent_id\": \"agent_a1b2c3d4e5f6...\",
    \"task_id\": \"$TASK_ID\",
    \"step\": 1,
    \"node_type\": \"START_NODE\",
    \"node_name\": \"start\",
    \"input_data\": {\"user_query\": \"What is the weather in Amsterdam?\"},
    \"output_data\": {},
    \"internal_data\": {},
    \"meta\": {},
    \"timestamp\": \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",
    \"is_external_call\": false,
    \"has_write_permission\": false
  }"

Response:

{
  "log_id": "log_abc123...",
  "agent_id": "agent_a1b2c3d4e5f6...",
  "task_id": "550e8400-e29b-41d4-a716-446655440000",
  "step": 1,
  "node_type": "START_NODE",
  "environment": "development",
  "created_at": "2026-01-15T10:00:00Z",
  "incidents": []
}

Step 2 — LLM_CALL

curl -X POST http://localhost:8000/api/v1/logs \
  -H "Authorization: Bearer KvKey-your-api-key" \
  -H "Content-Type: application/json" \
  -d "{
    \"agent_id\": \"agent_a1b2c3d4e5f6...\",
    \"task_id\": \"$TASK_ID\",
    \"step\": 2,
    \"node_type\": \"LLM_CALL\",
    \"node_name\": \"generate_response\",
    \"input_data\": {\"prompt\": \"What is the weather in Amsterdam?\"},
    \"output_data\": {\"response\": \"The weather in Amsterdam is 12°C and cloudy.\"},
    \"internal_data\": {\"model\": \"gpt-4o-mini\", \"tokens\": 42},
    \"meta\": {\"duration_ms\": 850},
    \"timestamp\": \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",
    \"is_external_call\": true,
    \"has_write_permission\": false
  }"

Step 3 — END_NODE

curl -X POST http://localhost:8000/api/v1/logs \
  -H "Authorization: Bearer KvKey-your-api-key" \
  -H "Content-Type: application/json" \
  -d "{
    \"agent_id\": \"agent_a1b2c3d4e5f6...\",
    \"task_id\": \"$TASK_ID\",
    \"step\": 3,
    \"node_type\": \"END_NODE\",
    \"node_name\": \"end\",
    \"input_data\": {},
    \"output_data\": {\"status\": \"completed\"},
    \"internal_data\": {},
    \"meta\": {},
    \"timestamp\": \"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",
    \"is_external_call\": false,
    \"has_write_permission\": false
  }"

---

What you're responsible for

When calling the API directly (without the SDK), a few things are your responsibility:

Concern	What to do
task_id	Generate a UUID at the start of each task run. Use the same value for all steps in that run.
step numbering	Start at 1, increment by 1 for each step. The API does not enforce ordering but the hash chain validation will fail if steps arrive out of sequence.
timestamp	Provide an ISO 8601 timestamp for when the step actually occurred.
START_NODE / END_NODE	Always begin with START_NODE and end with END_NODE. A task without an END_NODE will be flagged as unfinished after a timeout.
is_external_call	Set to true if this step calls an external API or service. Used by policies that restrict external calls.
has_write_permission	Set to true if this step can modify data (write to a database, send an email, etc.). Used by policies that restrict write operations.

The hash chain is computed server-side — you don't need to calculate or pass hashes yourself. The server chains each incoming log entry to the previous one for that task_id.

---

A minimal Python example

The same logic in Python without the SDK:

import uuid
import requests
from datetime import datetime, timezone

API_URL   = "http://localhost:8000"
API_KEY   = "KvKey-your-api-key"
AGENT_ID  = "agent_a1b2c3d4e5f6..."  # from registration response

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

def log_step(task_id: str, step: int, node_type: str, node_name: str,
             input_data: dict = None, output_data: dict = None, **kwargs):
    payload = {
        "agent_id":   AGENT_ID,
        "task_id":    task_id,
        "step":       step,
        "node_type":  node_type,
        "node_name":  node_name,
        "input_data":    input_data or {},
        "output_data":   output_data or {},
        "internal_data": kwargs.get("internal_data", {}),
        "meta":          kwargs.get("meta", {}),
        "timestamp":     datetime.now(timezone.utc).isoformat(),
        "is_external_call":    kwargs.get("is_external_call", False),
        "has_write_permission": kwargs.get("has_write_permission", False),
    }
    response = requests.post(f"{API_URL}/api/v1/logs", json=payload, headers=headers)
    response.raise_for_status()
    return response.json()


# Run a task
task_id = str(uuid.uuid4())

log_step(task_id, 1, "START_NODE", "start",
         input_data={"query": "What is the weather in Amsterdam?"})

log_step(task_id, 2, "LLM_CALL", "generate_response",
         input_data={"prompt": "What is the weather in Amsterdam?"},
         output_data={"response": "The weather in Amsterdam is 12°C and cloudy."},
         internal_data={"model": "gpt-4o-mini"},
         meta={"duration_ms": 850},
         is_external_call=True)

log_step(task_id, 3, "END_NODE", "end",
         output_data={"status": "completed"})

print(f"Task {task_id} logged successfully.")

---

Checking your work

After running the above, you can retrieve the full task log:

curl http://localhost:8000/api/v1/logs/task/$TASK_ID \
  -H "Authorization: Bearer KvKey-your-api-key"

And validate the hash chain:

curl http://localhost:8000/api/v1/logs/validate/$TASK_ID \
  -H "Authorization: Bearer KvKey-your-api-key"
# → {"valid": true, "total_steps": 3, "validated_steps": 3, "errors": []}

Both are also visible in the dashboard under Logs.

---

When to use the SDK instead

The REST API is the right choice when you're working outside Python, integrating from a language-native framework, or building a thin integration layer yourself. If you're writing a Python agent, the SDK's @kv.log_step decorator handles task_id generation, step numbering, timestamping, and error handling automatically — and adds one line of code per step rather than ~10. See Custom Python Agent.