omai-client Python SDK

The Omai Python SDK currently provides interfaces for Open-Vocabulary Object Detection (OD), OmChat, and VLX:

• od: Open-Vocabulary Object Detection
• om_chat: Synchronous multimodal OmChat requests
• vlx_seek: Synchronous multimodal VLX-Seek requests
• vlx_flow_stream: Create VLX streaming tasks and consume raw SSE events
• vlx_flow_stream_message: Create VLX streaming tasks and consume business messages
• vlx_flow_stream_raw: Return the raw streaming response
• vlx_flow_stop: Stop a VLX streaming task
• vlx_flow_update_config: Update the configuration of a VLX streaming task

Install from tar.gz

pip install ./dist/omai-1.0.3.tar.gz

Single-File Verification

For temporary verification, copy dist_single/vlx_api_client.py directly into your application project:

from vlx_api_client import VlxApiClient, VlxSeekRequest, VlxSeekMessage, VlxSeekContent

The single-file and package versions share the same core code. Real HTTP requests require requests:

pip install requests

Environment Variables

The Python SDK is configured through environment variables:

export VLX_API_ENABLED=true
export VLX_API_ENDPOINT=http://linker-gateway-service:31003/
export VLX_API_KEY=aaas_xxx
export VLX_API_CONNECT_TIMEOUT_MS=60000
export VLX_API_READ_TIMEOUT_MS=60000
export VLX_API_WRITE_TIMEOUT_MS=60000
export VLX_API_STREAM_READ_TIMEOUT_MS=0
export VLX_API_CLIENT_NUM=2048

VLX_API_STREAM_READ_TIMEOUT_MS=0 disables the read timeout for streaming interfaces and is suitable for 12- or 24-hour SSE connections.

Open-Vocabulary Object Detection Example

from omai_client import OdRequest, OdSubItem, VlxApiClient

client = VlxApiClient.from_env()

response = client.od(
    OdRequest(
        subList=[
            OdSubItem(
                model="OD210_021_002111_008",
                configCode="1646130545656070144",
            )
        ],
        input="https://example.com/demo.jpg",
        inputType="image_url",
        videoCode="videoCode_demo",
    )
)

for result in response.data:
    print(result.model, result.bboxList)

OmChat Example

from omai_client import (
    OmChatContent,
    OmChatImageUrl,
    OmChatMessage,
    OmChatRequest,
    VlxApiClient,
)

client = VlxApiClient.from_env()

response = client.om_chat(
    OmChatRequest(
        model="OmChat",
        messages=[
            OmChatMessage(
                role="user",
                content=[
                    OmChatContent(type="text", text="Describe the image content"),
                    OmChatContent(
                        type="image_url",
                        image_url=OmChatImageUrl(url="https://example.com/demo.jpg"),
                    ),
                ],
            )
        ],
        max_tokens=512,
        temperature=0.2,
    )
)

print(response.choices[0].message.content)

VLX-Seek Example

from omai_client import (
    VlxApiClient,
    VlxSeekContent,
    VlxSeekImageUrl,
    VlxSeekMessage,
    VlxSeekRequest,
)

client = VlxApiClient.from_env()

request = VlxSeekRequest(
    model="VLX-Seek",
    messages=[
        VlxSeekMessage(
            role="user",
            content=[
                VlxSeekContent(type="text", text="Describe the image content"),
                VlxSeekContent(
                    type="image_url",
                    image_url=VlxSeekImageUrl(url="https://example.com/a.jpg"),
                ),
            ],
        )
    ],
    max_tokens=512,
    temperature=0.2,
)

response = client.vlx_seek(request)
print(response.choices[0].message.content)

Streaming Message Example

from omai_client import VlxApiClient, VlxFlowSource, VlxFlowStreamRequest


class Handler:
    def on_event(self, message):
        print(message.event, message.content)

    def on_error(self, throwable):
        print("error:", throwable)

    def on_closed(self):
        print("closed")


client = VlxApiClient.from_env()
call = client.vlx_flow_stream_message(
    VlxFlowStreamRequest(
        source=VlxFlowSource(path="rtsp://camera", type=1),
        prompt="Identify abnormal situations in the scene",
        frame_preempt="qa",
        model="VLX-Stream",
    ),
    Handler(),
)

# Call this when the local SSE connection needs to be stopped.
# call.cancel()