Quickstart Guide

export KOLENA_API_KEY="your-api-key"

from kolena_agents import Client

client = Client()

# add new Agent Run
new_run = client.agent_run.add(agent_id=1, files=["path/to/file1", "path/to/file2"])

# each entry may be a path, an open binary file object, bytes, or a (filename, bytes_or_file_object) tuple
with open("path/to/large_file", "rb") as f:
    streamed_run = client.agent_run.add(agent_id=1, files=[f])

# download Agent Run
run = client.agent_run.get(agent_id=1, run_id=2)

# alternatively, list all Agent Runs
all_runs = client.agent_run.list(agent_id=1)

# delete Agent Run
client.agent_run.delete(agent_id=1, run_id=2)

# list all Agents
all_agents = client.agent.list()

# get an Agent
agent = client.agent.get(agent_id=1)

# update Agent properties
updated_agent = client.agent.update(
    agent_id=1,
    metadata={
        "environment": "production",
        "version": 2,
        "tags": ["nlp", "extraction"]
    }
)

# update Agent properties
agent = client.agent.update(
    agent_id=1,
    metadata={
        "environment": "production",
        "version": 2,
        "department": "finance",
        "tags": ["invoice-processing", "high-priority"],
        "threshold": 0.85,
        "enabled": True
    }
)

# access metadata from retrieved agent
agent = client.agent.get(agent_id=1)
print(agent.metadata)

# clear metadata
client.agent.update(agent_id=1, metadata={})

# add new Agent Run and wait for processing to complete
new_run = client.agent_run.add_wait(agent_id=1, files=["path/to/file1"])

# if the Run times out (still in "running" status), fall back to polling:
while new_run.status == "running":
    new_run = client.agent_run.get(agent_id=1, run_id=new_run.run_id)
    time.sleep(10)

agent = client.agent.get(agent_id=...)
output_template_id = agent.output_templates[0].id

# download Agent Run output as PDF
client.agent_run.download_templated_output(
    agent_id=...,
    run_id=...,
    output_template_id=output_template_id,
    output_format="pdf",
    filename="output.pdf",
)

time_range = {"start_time": "2025-09-09T00:00:00Z", "end_time": "2025-09-10T23:59:59Z"}

# initial query
response = client.access_logs.query(
    time_range=time_range,
    page_size=100
)
all_events = response.events

# fetch remaining pages
while response.next_cursor:
    response = client.access_logs.query(
        time_range=time_range,
        page_size=100,
        cursor=response.next_cursor
    )
    all_events.extend(response.events)

# list usage records
usage = client.usage.list(page_size=50)

# filter by date range
import datetime
now = datetime.datetime.now(datetime.timezone.utc)
usage = client.usage.list(
    created_gte=now - datetime.timedelta(days=30),
    created_lte=now,
)

# filter by agent or user
usage = client.usage.list(agent_id=1)
usage = client.usage.list(user="user@example.com")

from datetime import datetime, timezone

new_key = client.api_key.create(name="New Key", expiration_days=10)
print(f"New key: {new_key.api_key}")  # save this immediately!

# monitor keys for expiration
utc_now = datetime.now(timezone.utc)
response = client.api_key.list(page_number=0, page_size=50)
for key in response.api_keys:
    if key.status == "active" and key.expires_at:
        days_until_expiry = (key.expires_at - utc_now).days
        if days_until_expiry < 30:
            print(f"Warning: {key.name} expires in {days_until_expiry} days")

key = client.api_key.get(key_id=new_key.id)
print(f"Key status: {key.status}")

# delete an API key
client.api_key.delete(key_id="old-key-id")

from kolena_agents import Client
from kolena_agents.types.agent_run import AgentRun

client = Client()
run = client.agent_run.get(agent_id=1, run_id=2)

run_dict = run.model_dump(mode="json")  # get the AgentRun as a dictionary
run = AgentRun.model_validate(run_dict)  # deserialize an AgentRun

from kolena_agents import webhook

result = webhook.construct_event(request_body, secret, request_headers)