ohttpy Package

Introduction

This package allows for client-side OHTTP-wrapping over HTTP communication. It acheives this by providing drop-in replacement classes for requests.Session and httpx.BaseTransport classes that abstract the OHTTP layer.

NOTE: This package utilizes a modified Rust library that implements OHTTP and relies on the chunked OHTTP protocol which has not been ratified into a standard at the time of writing.

Optional Parameter Configurability

The ohttpy.Transport and ohttpy.Session class constructors take in the following optional parameters:

• key_config: The OHTTP Key Configuration and its encoding format are defined by OHTTP RFC9458. This configuration contains the OHTTP public key and the set of algorithms for the HPKE context. By default, the classes will fetch the OHTTP Key Configuration dynamically from {url.scheme}://{url.authority}/discover where url is the OHTTP Gateway address. This is not the usual flow for OHTTP clients so this parameter allows the user to provide the key configuration.

Requests Session Example

To use OHTTP in a requests.Session setting, simply replace the object construction with ohttpy.Session:

import ohttpy

# Create ohttpy session that inherits from requests.Session.
session = ohttpy.Session()

# Use session like a request.Session including streaming response support.
resp = session.post("http://ohttp-server/path/to/post", data="test data", stream=True)
resp.raise_for_error()
for chunk in resp.iter_content():
    if chunk:
        print(chunk.decode(), end="", flush=True)

HTTPX Transport Example

To use OHTTP in a httpx.Transport setting, simply replace the object construction with ohttpy.Transport:

import httpx
import ohttpy

# Initialize httpx client with the ohttpy Transport that inherits from httpx.Transport
httpx_client = httpx.Client(transport=ohttpy.Transport())

# Use client as normal including streaming response support.
method = "GET"
url = "http://ohttp-server/path/to/get/"
with httpx_client.stream(method, url) as resp:
    resp.raise_for_error()
    for chunk in resp.stream:
        print(chunk.decode(), end="", flush=True)

This is also applicable to clients that utilize httpx for their HTTP communication, for example openai client:

import httpx
import openai
import ohttpy

# Initialize httpx client with the ohttpy Transport that inherits from httpx.Transport
httpx_client = httpx.Client(transport=ohttpy.Transport())
url = "http://ohttp-server"
api_key = "API_KEY"

# Configure OpenAI client with httpx client
client = openai.OpenAI(
    api_key=api_key,
    http_client=httpx_client,
    base_url=url + "/v1")

# Use OpenAI SDK as normal for example llama chat (including stream capability)
completion = client.chat.completions.create(
    model="meta-llama/Llama-3.2-3B-Instruct",
    messages=[
        {"role": "system", "content": "You are a helpful AI assistant."},
        {"role": "user", "content": "where does the sun rise from?"},
    ],
    temperature=0.2,
    top_p=0.7,
    max_tokens=1024,
    stream=True,
)
for chunk in completion:
    print(chunk.choices[0].delta.content or "", end="", flush=True)