ai-stream-proxy-sdk 0.2.1

Sync and async Python SDK for the Pantheon stream-proxy service.

ai-stream-proxy-sdk

Python SDK for the Pantheon stream-proxy service with aligned sync and async APIs.

Sync

from ai_stream_proxy import StreamProxyClient

with StreamProxyClient(server_url="http://127.0.0.1:3000") as client:
    client.create_stream(
        stream_id="example-sync",
        message_id="msg-sync",
        content_type="claude-code-stream-json+include-partial-messages",
    )

    with client.new_stream(stream_id="example-sync") as stream:
        stream.enqueue(b'{"type":"thread.started","thread_id":"thread-1"}\n')

Async

import asyncio

from ai_stream_proxy import AsyncStreamProxyClient


async def main() -> None:
    async with AsyncStreamProxyClient(server_url="http://127.0.0.1:3000") as client:
        await client.create_stream(
            stream_id="example-async",
            message_id="msg-async",
            content_type="codex-stream-json",
        )

        async with client.new_stream(stream_id="example-async") as stream:
            stream.enqueue(b'{"type":"thread.started","thread_id":"thread-1"}\n')


asyncio.run(main())

Delayed Stop Signal

If another service should send the terminal stop signal later, disable auto-finish on context exit:

async with client.new_stream(
    stream_id="example-async",
    auto_finish_on_exit=False,
) as stream:
    stream.enqueue(b"chunk")
    await stream.close_without_stop()