pumpsdk 0.2.0

Drop-in OpenAI-compatible Python SDK for the Pump LLM gateway

pump Python SDK

A drop-in replacement for the official OpenAI Python SDK for pump — a multi-tenant LLM gateway that exposes an OpenAI-compatible API.

Keep your existing OpenAI request code exactly as-is. The only things that change are the import, the base_url, and using your pump key (pk-...) as the API key.

Installation

pip install pumpsdk

The distribution is published as pumpsdk; the import package is pump.

Quickstart — OpenAI drop-in

# before:
# from openai import OpenAI
# client = OpenAI()

from pump.openai import OpenAI

client = OpenAI(
    base_url="https://api.pump.co/ai/v1",
    api_key="pk-...",  # your pump key (or set PUMP_API_KEY)
)

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)

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

That's the whole migration. Every standard parameter (temperature, tools, tool_choice, response_format, seed, stream, ...) is forwarded to the gateway unchanged.

Pump is the same client under a more obvious name — from pump import Pump and from pump.openai import OpenAI are interchangeable.

API key

Pass api_key="pk-..." explicitly, or omit it and the SDK reads PUMP_API_KEY from the environment:

export PUMP_API_KEY="pk-..."

from pump.openai import OpenAI

client = OpenAI()  # base_url defaults to https://api.pump.co/ai/v1

Responses & embeddings

client.responses.create(model="gpt-4o-mini", input="Write a haiku about pumps.")

client.embeddings.create(model="text-embedding-3-small", input="hello world")

Streaming

The streamed object is iterable directly — no context manager required:

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Count to 5"}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta
    if getattr(delta, "content", None):
        print(delta.content, end="", flush=True)

You can also use it as a context manager to guarantee the connection closes:

with client.chat.completions.create(..., stream=True) as stream:
    for chunk in stream:
        ...

Async

import asyncio
from pump.openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI(api_key="pk-...")

    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(resp.choices[0].message.content)

    # async streaming
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Count to 5"}],
        stream=True,
    )
    async for chunk in stream:
        ...

    await client.aclose()

asyncio.run(main())

Attaching metadata & tags (pump extension)

pump lets you attach observability metadata and tags to any request. These are not part of the OpenAI request body — sending custom fields there would be rejected by the upstream provider. Instead, pass them as explicit kwargs and the SDK sends them as gateway headers (x-pump-metadata, x-pump-tags), which the gateway reads and strips before forwarding the request upstream.

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
    metadata={"user_id": "u_123", "feature": "support-bot"},
    tags=["prod", "support"],
)

You can also pass arbitrary headers via extra_headers:

client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    extra_headers={"x-pump-trace-id": "abc123"},
)

Note: OpenAI's own body-level metadata field (for stored completions) is a different thing. If you need it, pass it through extra_body={"metadata": {...}}.

Response objects

Responses support both attribute access and dict access, and expose .model_dump() to get a plain dict:

resp = client.chat.completions.create(...)

resp.choices[0].message.content        # attribute access
resp["choices"][0]["message"]["content"]  # dict access
resp.model_dump()                       # plain dict

Error handling

The SDK raises a single error type, pump.PumpError. For HTTP failures it carries the status_code (and response) so you can branch on it:

from pump import PumpError

try:
    client.chat.completions.create(...)
except PumpError as e:
    if e.status_code == 429:
        ...  # rate limited
    elif e.status_code == 401:
        ...  # bad API key
    else:
        raise

Roadmap: more providers

The SDK is built on a shared transport so additional provider shims can be added without changing existing code. An Anthropic-compatible shim (from pump.anthropic import Anthropic) is reserved as an extension point and will land once the gateway exposes an Anthropic-compatible endpoint.

Development

pip install -e ".[dev]"
pytest

Tests use httpx.MockTransport and never call the real API.

License

MIT