lowcoai-agentx

Python client for the agentx backend: the agent-manager, agent-kb (knowledge-base) and agent-executor services in one package. Sync and asyncio clients with the same surface, built on httpx. Python 3.10+.

pip install lowcoai-agentx

Quick start

from lowcoai.agentx import AgentxClient, MessageSendParams

params: MessageSendParams = {
    "message": {
        "role": "user",
        "messageId": "msg_001",
        "kind": "message",
        "parts": [{"kind": "text", "text": "Hello!"}],
    },
    "chatType": "chat",
}

with AgentxClient("<token-or-api-key>", org_id="org_123") as client:
    agents = client.manager.list_agents(page_no=1, size=25)
    kbs = client.kb.list_knowledge_bases()

    resp = client.executor.send_message("agent_abc", params)
    if "error" in resp:
        print("rpc error", resp["error"])
    else:
        print("result", resp.get("result"))

The three sub-clients (client.manager, client.kb, client.executor) share one connection pool and one set of default headers.

asyncio

from lowcoai.agentx import AsyncAgentxClient

async with AsyncAgentxClient("<token-or-api-key>", org_id="org_123") as client:
    agents = await client.manager.list_agents()

AsyncAgentxClient has the same sub-clients and methods as AgentxClient; each endpoint method is a coroutine (stream_message is an async generator). Close it with await client.aclose() if you don't use async with.

Streaming the executor

from lowcoai.agentx import try_parse_message

for ev in client.executor.stream_message("agent_abc", params):
    msg = try_parse_message(ev)
    if msg:
        print("delta:", msg)
    else:
        print("raw:", ev.data)

Each StreamEvent carries the raw data of one server-sent event (multi-line data: fields joined with "\n"); try_parse_message(ev) decodes it as JSON and returns None when it is not a JSON object. The request is sent when iteration starts, a non-2xx answer raises AgentxError, and the read timeout does not apply to the stream. Breaking out of the loop closes the connection; with asyncio, wrap the generator in contextlib.aclosing to close it deterministically:

from contextlib import aclosing

async with aclosing(client.executor.stream_message("agent_abc", params)) as events:
    async for ev in events:
        if ev.data == "[DONE]":
            break

Options

All requests go to https://api.lowco.ai.

client.set_org_id(...) and client.set_header(key, value) change headers for later requests of all three sub-clients; pass None to remove.

Data shapes

Request and response bodies are plain dicts typed as TypedDicts (Agent, LlmModel, Conversation, KnowledgeBase, Dataset, A2AMessage, MessageSendParams, JSONRPCResponse, …) whose keys match the JSON on the wire (camelCase). A value returned by one call can be sent back as the body of another. Message parts are the TextPart / FilePart / DataPart union (Part), discriminated by kind.

List methods take keyword-only page_no, size, filter and sort. The get_*_count methods return a plain int.

Constants: HEADER_ORG_ID, DEFAULT_MANAGER_API_BASE_PATH, DEFAULT_KB_API_BASE_PATH, DEFAULT_EXECUTOR_API_BASE_PATH, METHOD_MESSAGE_SEND, JSONRPC_ERROR_CODES, A2A_ERROR_CODES.

Surface

health calls GET /health at the host root.

Errors

Every HTTP-level non-2xx response raises AgentxError with status_code, message, code and the raw body. JSON-RPC level errors from the executor are not raised: they come back in the error field of the JSONRPCResponse.

from lowcoai.agentx import AgentxError

try:
    client.manager.get_agent("missing")
except AgentxError as err:
    print(err.status_code, err.code, err.message, err.body)

message and code come from the service's error envelope when present, otherwise from a top-level message, otherwise message is the raw response text. Network failures and timeouts surface as httpx exceptions.