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)

A bot can work with each person's Microsoft 365 when a Microsoft connector has the microsoft365 feature. Areas are mail, calendar, files (read or read_write), people, and meetings (read):

client.bots.update(bot.id, microsoft365={"mail": "read_write", "calendar": "read"})

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})
client.model_settings.update(assistant_model="own-large")  # the Console assistant

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

MCP connectors

An MCP connector is an org entry for an MCP server. Bots name it with {"type": "mcp", "connector_id": connector.id}. The auth mode is none, shared (a credential_id of an org vault), or per_person: each person saves a token, or signs in at the OAuth app.

jira = client.mcp_connectors.create(
    label="jira", url="https://jira.example/mcp", auth_mode="per_person"
)
client.mcp_connectors.put_token(jira.id, "my-token")  # the caller's own, write-only
found = client.mcp_connectors.probe(url="https://crm.example/mcp")  # OAuth, and which client
crm = client.mcp_connectors.create(
    label="crm",
    url="https://crm.example/mcp",
    auth_mode="per_person",
    oauth={"mode": "discover"},
)
start = client.mcp_connectors.connect(crm.id, return_url="https://chat.example/settings")
offered = client.mcp_connectors.tools(jira.id)  # names, titles, and the server's hints
client.bots.update(bot.id, tools=[{"type": "mcp", "connector_id": jira.id}])

Skill credentials

A bot maps each credential name its skills declare to an OAuth app (each person's own account) or to an org vault credential:

client.bots.update(
    bot.id,
    skill_credentials={
        "GITHUB_TOKEN": {"oauth_app_id": app.id},
        "CRM_KEY": {"credential_id": credential.id},
    },
)

OAuth apps and connected accounts

An OAuth app is an org object for any OAuth provider. A connected account is one person's sign-in. The client secret is write-only, and no response holds a token.

app = client.oauth_apps.create(
    name="CRM",
    authorize_url="https://crm.example/oauth/authorize",
    token_url="https://crm.example/oauth/token",
    client_id="…",
    client_secret="…",
    scopes=["read"],
)
start = client.oauth_apps.connect(app.id, return_url="https://chat.example/settings")
print(start.authorize_url)  # send the browser here
for account in client.connected_accounts.iter():
    print(account.provider_name, account.label, account.status)
client.connected_accounts.disconnect(account.id)

connected_accounts.list(all=True) lists every account of the org for a key with connector:write. oauth_apps.client_credentials(app.id, vault_id=...) gets a token for an app with the client_credentials grant.

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)

The feature microsoft365 gives bots each person's mail, calendar, files, people, and meetings. It needs delegated Graph permissions that the tenant admin consents to once:

setup = client.connectors.microsoft365(teams.id)
print(setup.scopes, setup.redirect_uri, setup.admin_consent_url, setup.missing_scopes)
start = client.connectors.connect_microsoft365(
    teams.id,
    return_url="https://chat.example/settings",
    prompt="none",
    login_hint="ada@acme.example",
)

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)

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.

Owners and admins read personal sessions of other people only after a takeover for 2 hours. Each takeover is a row of an audit log:

client.session_takeovers.create()  # a running takeover is returned, not added again
for row in client.session_takeovers.iter(active=True):
    print(row.subject_id, row.until)

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:

Class	When
BadRequest	400, 413 and other 4xx
AuthError	401, 403
BudgetExhausted	402
NotFound	404
Conflict	409; conflicts lists PushConflict(path, base_sha256, current_sha256) for a Knowledge Base push_conflict
AcknowledgementRequired	409 acknowledgement_required; entries lists what to acknowledge
RateLimited	429; retry_after in seconds
ServerError	5xx
APIConnectionError, APITimeoutError	no response

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"})
assistant = client.sessions.create(assistant={})  # the Console assistant
client.sessions.message(assistant.id, "Shorten the purpose.", metadata={"page": "bot-editor"})
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). Knowledge Bases belong to the organization. Your own memory is separate (below).

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)
recent = client.knowledge_bases.pages(kb.id, sort="updated", limit=50)  # a window; also q, folders
detail = client.knowledge_bases.revision(kb.id, 2, paths=["travel.md"])  # diffs for these paths
chat = client.knowledge_bases.open_curator(kb.id)  # a session with the built-in curator

Memory

Every person has one memory per organization. The Personal bot reads and writes it; company bots can read it. The calls work like the Knowledge Base calls and always address your own memory. Owners and admins read and change the settings of every memory and see numbers only.

mine = client.memory.get()  # revision, size, limits
pages = client.memory.pages()
client.memory.push(
    message="Prefers short answers",
    changes=[{"op": "put", "path": "preferences.md", "content": "...", "base_sha256": None}],
)
client.memory.clear()  # your pages and history
stats = client.memory.settings().stats  # people, pages, bytes
client.memory.update_settings(limits={"max_bytes": 131072})

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 skills version SKILL skill.zip    # a new version; bots on an older version keep it

geki triggers create BOT --kind cron --cron "0 9 * * 1" --timezone Europe/Berlin --prompt "Weekly report"
geki triggers create BOT --kind poll -f poll.yaml   # poll: {source: mail, sender: anna}
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. geki memory show|clear|settings shows and clears your memory and, for owners and admins, changes the settings of every memory. 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