geki

Python client for the GEKI API.

pip install geki

Python 3.12 or newer.

Stability

The GEKI API has three levels. The model API (/v1/models, /v1/chat/completions, /v1/embeddings) is stable. The bots API (bots, templates, triggers, runs) is beta and will become stable. Everything else (agents, sessions, vaults, files, uploads, skills, knowledge bases) is beta and may change without a transition. This package stays 0.x while the bots API is beta. Update it when a beta call breaks.

For the model API you can also use the OpenAI SDK with base_url="https://api.geki.ai/v1".

Auth

Create an API key in the GEKI Console. Pass it or set GEKI_API_KEY. GEKI_API_URL changes the base URL (default https://api.geki.ai/v1).

from geki import Geki

client = Geki()  # reads GEKI_API_KEY

Key, health and usage

me = client.me()  # org_id, subject_id, subject_kind, permissions of this key
client.health()  # {"status": "ok"}, without the key

report = client.usage.query(
    start="2026-10-01T00:00:00Z", end="2026-11-01T00:00:00Z", groupings=[["model"], ["day"]]
)
print(report.totals.billed_eur)
for row in client.usage.requests(model="glm-5.3-flash", limit=20).data:
    print(row.ts, row.input_tokens, row.output_tokens)

Usage covers the key's org and needs usage:read.

Extra headers

with_headers returns a client that sends the headers on every call. It shares the connections of the client it came from.

tagged = client.with_headers({"x-request-id": "nightly-42"})
tagged.chat.completions.create(model="glm-5.3-flash", messages=[{"role": "user", "content": "Hi"}])

Bots

bot = client.bots.create(name="Invoices", model="glm-5.3-flash", instructions="Sort invoices.")
client.bots.update(bot.id, tags=["finance"])

caps = client.bots.capabilities(bot.id)
client.bots.publish(
    bot.id,
    acknowledgements=[{"id": e.id, "fingerprint": e.fingerprint} for e in caps.entries],
)

for row in client.bots.iter(status="live"):
    print(row.id, row.name)

Triggers and runs:

trigger = client.triggers.create(bot.id, kind="on_demand", input_prompt="Summarize today.")
run = client.triggers.fire(bot.id, trigger.id, input="Only open tickets.")
print(client.runs.get(run.id).status)

An on-demand trigger returns a token and a Standard Webhooks secret once. Send to its hook with either:

hook = client.triggers.create(bot.id, kind="on_demand", input_prompt="New ticket:")
client.hooks.send(hook.id, {"ticket": 42}, secret=hook.secret)
client.hooks.send(hook.id, {"ticket": 43}, token=hook.token)

geki.webhooks.sign(secret, body) returns the Standard Webhooks headers if you send the request yourself.

Model settings and BYOK endpoints

The org picks where tokens come from. model="default" is the org default model. Changes need models:write. A BYOK endpoint is an OpenAI-compatible endpoint with your own provider key; the key is write-only.

print(client.model_settings.get().default_model)
endpoint = client.byok_endpoints.create(
    name="Own provider",
    base_url="https://llm.example.com/v1",
    key="sk-…",
    models=[
        {
            "id": "own-large",
            "provider_model": "large-2",
            "context_window": 128000,
            "max_tokens": 8192,
        }
    ],
)
print(client.byok_endpoints.test(endpoint.id).ok)
client.model_settings.update(default_model="own-large", geki_enabled=False)
client.model_settings.update({"geki_enabled": None})

None in a mapping removes the org's own value, so it inherits again.

Connectors and channels

A connector is a Slack app or a Microsoft (Teams) app of your org. Credentials are write-only.

slack = client.connectors.create(
    provider="slack",
    name="Acme Slack",
    credentials={"bot_token": "xoxb-…", "signing_secret": "…"},
    settings={"bot_ids": [bot.id], "entry_bot_id": bot.id},
)
app = client.connectors.manifest(slack.id, name="Acme Bots")
client.connectors.update(slack.id, manifest_version=app.version)
print(client.connectors.test(slack.id).status)
for channel in client.connectors.channels(slack.id):
    print(channel.id, channel.name, channel.default_bot_id)
client.connectors.set_channel(slack.id, "C123", default_bot_id=bot.id)

A Microsoft connector has features. teams_package returns the Teams app zip:

teams = client.connectors.create(
    provider="microsoft",
    name="Acme Teams",
    credentials={
        "tenant_id": "…",
        "app_id": "…",
        "client_secret": "…",
        "secret_expires_at": "2027-04-01",
    },
    features=["teams", "teams_files"],
)
package = client.connectors.teams_package(
    teams.id,
    name="Acme Bots",
    publisher={
        "name": "MSP GmbH",
        "website": "https://msp.example",
        "privacy_url": "https://msp.example/privacy",
        "terms_url": "https://msp.example/terms",
    },
)
open("acme-bots.zip", "wb").write(package.content)

A trigger posts its answer to channels with outputs:

client.triggers.update(
    bot.id,
    trigger.id,
    outputs=[{"connector_id": slack.id, "channel_id": "C123"}],
    notify_on_failure=True,
)

Setting or changing outputs needs connector:write. iter_all lists the triggers of the org, here the ones that post through one connector:

posting = list(client.triggers.iter_all(connector_id=slack.id))

Personal bot

Each org has one Personal bot. Members tune it within what the org allows.

slot = client.personal_bot.create()  # get() reads without creating
client.personal_bot.put_settings(instructions="Answer in German.", web_search=True)
client.personal_bot.put_token("jira", "my-token")  # write-only; delete_token() removes it

session = client.sessions.create(
    agent_id=slot.agent_id, personal_options={"workspace": False, "web_search": True}
)
client.sessions.update(session.id, personal_options={"web_search": False})

The owner edits the slot like any bot, with member_options in the draft and owner_can_read as bot state: client.bots.update(slot.bot_id, owner_can_read=True).

Responses

Every call returns a Pydantic model from geki.models. Fields the client does not know yet stay available in model_extra. to_dict() returns plain JSON.

Lists return one page (data, has_more, last_id). iter() walks all pages:

page = client.runs.list(bot_id=bot.id, limit=20)
for run in client.runs.iter(bot_id=bot.id, max_pages=5):
    ...

client.request(method, path, ...) calls any route and returns the JSON.

Errors

Errors raise geki.GekiError with status, code, message, param, fields, entries and request_id. Branch on the class or on code:

Embeddings

result = client.embeddings.create(model="qwen3-embedding-8b", input=["first", "second"])
vectors = [item.embedding for item in result.data]

Streaming

Session events and chat completions stream as server-sent events. send posts one input event and returns when the API accepted it (202). send_stream streams that turn until it ends, message waits for the turn and returns the session. stream_events follows the session and stays open; events lists the stored events as JSON. Event fields are at the top level, as at OpenAI (event_id, type, delta, …); a failure is the event error with error.code and error.message. Timestamps of sessions, turns, items, artifacts, agents and vaults are Unix seconds.

turns, items and artifacts return one page, newest first. iter_turns, iter_items and iter_artifacts walk the whole session, oldest first. A message item has role and content, a list of parts with type and text.

from geki import message_event

session = client.sessions.create(agent_id=agent.id, environment={"type": "none"})
with client.sessions.send_stream(session.id, message_event("Hello")) as stream:
    for event in stream:
        print(event.event, event.json())

preview = client.sessions.create(preview={"bot_id": bot.id, "state": "draft"})
current = client.sessions.list(preview_bot_id=bot.id, preview_state="draft", current_preview=True)

for chunk in client.chat.completions.stream(
    model="glm-5.3-flash", messages=[{"role": "user", "content": "Hi"}]
):
    print(chunk["choices"][0]["delta"].get("content", ""), end="")

Files and skills

client.files.create("report.pdf", purpose="user_data")
content = client.files.content(file_id)  # Download(content, content_type, filename)
client.skills.create("skill.zip", name="Mail")
client.uploads.upload("big.csv", purpose="file")  # presigned upload for large files

Knowledge Bases

Pages are text files that bots read and maintain. Each push is one revision and names the sha each change is based on (None for a new page). Use instance="me" for your own instance of a per-person Knowledge Base, such as memory.

kb = client.knowledge_bases.create(name="Wiki", template="company_wiki")
page = client.knowledge_bases.page(kb.id, "overview.md")
client.knowledge_bases.push(
    kb.id,
    message="Add the travel policy",
    changes=[{"op": "put", "path": "travel.md", "content": "...", "base_sha256": None}],
)
manifest, etag = client.knowledge_bases.manifest(kb.id)
same, _ = client.knowledge_bases.manifest(kb.id, etag=etag)  # None when unchanged
old = client.knowledge_bases.page(kb.id, "overview.md", revision=1)
client.knowledge_bases.restore(kb.id, "overview.md", revision=1)
mine = client.knowledge_bases.pages(memory_id, instance="me")
client.knowledge_bases.delete_instance(memory_id)  # your pages and history
chat = client.knowledge_bases.open_curator(kb.id)  # a session with the built-in curator

Async

AsyncGeki has the same transport (request, stream, paginate, download), with_headers, health, me, and the calls a chat UI needs: models, chat completions, embeddings, bots get, the Personal bot, sessions, uploads and file content.

from geki import AsyncGeki

async with AsyncGeki() as client:
    bot = await client.bots.get(bot_id)

Key refresh

on_unauthorized is called once after a 401. Return a new key to retry the request, or None to raise AuthError.

client = Geki(api_key, on_unauthorized=lambda: fetch_new_key())

CLI

The geki command ships as the extra geki[cli] (Typer, Rich, PyYAML):

uv tool install "geki[cli]"   # or: pipx install "geki[cli]"
geki auth login               # asks for the API key, checks it, saves it

The key is saved in ~/.config/geki/config.toml (mode 600) under a profile. Flags win over the environment, the environment over the profile: --api-key > GEKI_API_KEY > profile, --api-url > GEKI_API_URL > profile

https://api.geki.ai/v1. --profile or GEKI_PROFILE picks the profile (default default).

geki bots list --status live
geki bots get BOT -o yaml > bot.yaml
geki bots update BOT -f bot.yaml       # sends only the fields that changed
geki bots edit BOT                     # the draft in $EDITOR
geki bots publish BOT                  # asks to acknowledge new capabilities

geki triggers create BOT --kind cron --cron "0 9 * * 1" --timezone Europe/Berlin --prompt "Weekly report"
geki runs watch "$(geki triggers fire BOT TRIGGER -q)"
geki runs list --bot BOT --status failed --all -o json

Knowledge Bases: geki knowledge list|get|create|delete manages them. The files are synced with the kb command, which geki kb ... runs with the key of your profile. It comes with the extra kb (Linux x86_64):

uv tool install "geki[cli,kb]"   # or: pip install "geki[cli,kb]"
geki knowledge create --name Wiki --template company_wiki
geki kb clone wiki wiki && cd wiki
geki kb push -m "Add the setup page"

Every command has --help. -o table|json|yaml sets the output; JSON and YAML print the API object. -q prints only ids. Lists take --limit, --after and --all. Errors go to stderr with the code and the request id; the exit code is 1 for API errors and 2 for usage errors. --debug logs each request without the key. Trigger secrets and tokens are printed once and never saved.

Admin

GekiAdmin calls the internal admin routes with a system key. They are not reachable on api.geki.ai; only GEKI services use it. Resources: keys, subjects, service_accounts, staff_access, orgs, chat, usage, credits, billing, models, nodes.

License

MIT