Python SDK / ZeliClient

ZeliClient#

The client holds your configuration and event handlers, and opens sessions. All constructor arguments are keyword-only.

from zeli import ZeliClient

Constructor#

ZeliClient(
    *,
    api_key: str | None = None,
    avatar_id: str | None = None,
    avatar_config: AvatarConfig | None = None,
    options: ClientOptions | None = None,
)

Methods#

connect(session_options=None)#

Opens a session. The return value is both awaitable and an async context manager:

async with client.connect() as session:      # recommended
    ...
 
session = await client.connect()              # manual lifetime

Optionally takes a SessionOptions.

on(event)#

Decorator that registers a handler for a ZeliEvent. Handlers may be sync or async.

@client.on(ZeliEvent.MESSAGE_RECEIVED)
async def handler(message): ...

add_listener(event, callback) / remove_listener(event, callback)#

Programmatic registration and removal. Callbacks may be sync or async.

get_message_history() -> list[Message]#

Returns a copy of the conversation transcript accumulated so far, so mutating it won't affect the client.

register_tool_call_handler(name, handler, *, timeout=None) -> Callable[[], None]#

Answers one of your custom tools. Returns the function that unregisters it. Register before connect(): the persona can call a tool on its first turn, and a call that finds no handler is refused at once.

@client.tool_call_handler("open_device_settings")
async def open_device_settings(args, call):
    await settings_panel.open(args["section"])
    return {"opened": True}

handler(args, call) may be a coroutine function, which runs as a task and is cancelled when the answer can no longer matter, or a plain function, which runs on a worker thread so a blocking call cannot stall the socket. What it returns is sent as the result. A raised ToolCallError is sent as a failed result carrying its message, and the persona says it could not be done. Any other exception is sent as the generic "The tool failed." and delivered to ZeliEvent.ERROR with the original as __cause__, so a driver's error text never reaches the model. A result may be at most 256 KB of JSON (MAX_TOOL_RESULT_BYTES); a larger one is answered as a failure. A plain function must poll call.is_cancelled(), because a thread cannot be cancelled. timeout, in seconds, gives up sooner than the server would. One handler per name; a second is refused with ConfigurationError. @client.tool_call_handler(name, timeout=None) is the decorator form.

Whatever a handler returns or raises is read by a language model as untrusted data. Never return a secret.

Properties#

pending_tool_calls -> int#

How many tool calls are waiting on a handler right now. Zero after a session closes: a call in flight at that moment is cancelled and reported on TOOL_CALL_FAILED with session_closed.

credential_is_live -> bool | None#

The server's last word on the credential this session's control socket is using. None before it has said anything, which is deliberately not the same answer as False: a page that treated "never asked" as "dead" would warn on every healthy session that has simply never needed a renewal. Every change also arrives as a CREDENTIAL_STATE event.

Example#

from zeli import ZeliClient, AvatarConfig, ClientOptions, ZeliEvent
 
client = ZeliClient(
    avatar_config=AvatarConfig(avatar_id="01-presenter-male__confident"),
    options=ClientOptions(server_url="http://your-server:8080"),
)
 
@client.on(ZeliEvent.MESSAGE_RECEIVED)
async def log(message):
    print(message.role.value, message.content)
 
async with client.connect() as session:
    await session.send_message("Hello!")
    await session.wait_until_closed()