terrakernel-odxproxyclient

Python client SDK for ODXProxy — a reverse proxy that fronts one or more Odoo instances behind a unified JSON-RPC 2.0 API. By Terrakernel.

Importable as terrakernel.odxproxyclient (PEP 420 namespace under terrakernel).

• Sync and async, sharing one core. Pick the one that matches your runtime.
• HTTP/2 + brotli/gzip by default via httpx.
• Typed exceptions for every error code in the spec.
• v2 sessions for Odoo 19+ (for_instance_v2): ODXProxy 0.9.0+ /v2, which reaches Odoo over JSON-2 with named arguments and typed Odoo errors. It's ready for Odoo 22, which removes the /jsonrpc endpoint the v1 sessions use.
• Framework-friendly: drop-in patterns for FastAPI and Flask, but no framework dep.
• orjson for fast JSON serialization.
• Python >=3.12 (tested on 3.12 and 3.14). Wire protocol fully specified in SYSTEM_ARCHITECTURE.md.

---

Install

pip install terrakernel-odxproxyclient
# or, with uv
uv add terrakernel-odxproxyclient

Then import as:

from terrakernel.odxproxyclient import ODXProxyClient, AsyncODXProxyClient, OdooInstance

---

Two API keys, never confuse them

ODXProxy uses two completely separate credentials. The client mirrors this — they live in different places, intentionally:

Key	Where it goes	What it is
Proxy x-api-key	On the client (api_key=... in the constructor)	Sent as the x-api-key HTTP header. Authenticates you with the proxy.
Odoo api_key	On OdooInstance(api_key=...)	Carried inside the JSON-RPC body. Authenticates the proxy with Odoo as a specific user.

---

Quick start — sync

from terrakernel.odxproxyclient import ODXProxyClient, OdooInstance

with ODXProxyClient(
    base_url="https://proxy.example.com",
    api_key="PROXY_X_API_KEY",
) as client:
    session = client.for_instance(
        url="https://erp.example.com",
        db="prod",
        user_id=2,
        api_key="ODOO_USER_API_KEY",
    )

    partners = session.search_read(
        "res.partner",
        params=[[["is_company", "=", True]]],
        keyword={"fields": ["id", "name", "email"], "limit": 100},
    )
    print(partners)

Quick start — async

import asyncio
from terrakernel.odxproxyclient import AsyncODXProxyClient

async def main() -> None:
    async with AsyncODXProxyClient(
        base_url="https://proxy.example.com",
        api_key="PROXY_X_API_KEY",
    ) as client:
        session = client.for_instance(
            url="https://erp.example.com",
            db="prod",
            user_id=2,
            api_key="ODOO_USER_API_KEY",
        )
        partners = await session.search_read(
            "res.partner",
            params=[[["is_company", "=", True]]],
            keyword={"fields": ["id", "name"], "limit": 100},
        )
        print(partners)

asyncio.run(main())

---

API reference

Client constructors

Sync and async clients share the same signature:

ODXProxyClient(
    base_url: str,
    api_key: str,            # the proxy's x-api-key
    *,
    default_timeout_secs: float = 15.0,
    http_client: httpx.Client | None = None,        # async: httpx.AsyncClient
    http2: bool = True,
)

If http_client is provided, the SDK uses it and will NOT close it on close()/aclose(). Otherwise the SDK owns an internal httpx.Client and closes it.

OdooInstance (frozen dataclass)

All fields are required:

OdooInstance(
    url: str,        # full Odoo URL, e.g. "https://erp.example.com"
    db: str,         # Odoo database name
    user_id: int,    # Odoo user id (not login)
    api_key: str,    # Odoo user's api_key — different from the proxy x-api-key
)

Session factories

Two equivalent ways to build a Session / AsyncSession:

# Canonical: build OdooInstance once, reuse across sessions.
session = client.session(OdooInstance(url=..., db=..., user_id=..., api_key=...))

# Shortcut for one-shot scripts.
session = client.for_instance(url=..., db=..., user_id=..., api_key=...)

Both return the same object. Sessions are cheap (no I/O on construction) — make as many as you need.

Action methods on Session / AsyncSession

All signatures are uniform. First positional arg is model_id (the Odoo model name, e.g. "res.partner"). params and keyword are keyword-only. params is forwarded to Odoo's execute_kw as positional args; keyword is forwarded as kwargs. The client never rewrites Odoo domains.

def search(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[int]
def search_count(  model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> int
def read(          model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[dict]
def fields_get(    model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> dict
def search_read(   model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> list[dict]
def create(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> int | list[int]
def write(         model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> bool
def unlink(        model_id: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> bool
def call_method(   model_id: str, fn_name: str, *, params=None, keyword=None, request_id=None, timeout_secs=None) -> JsonValue

Return types are not statically narrowed. The methods' declared return type is JsonValue (None | bool | int | float | str | list | dict) because the proxy passes through whatever Odoo returns. The annotations above reflect what Odoo actually returns for each action — treat them as a guide, not a guarantee. If you need narrowing, assert isinstance(...) or cast(...) at the call site.

• request_id= — optional client-supplied UUID; if omitted, the SDK generates a UUID4.
• timeout_secs= — per-call only. Sent as the proxy's x-request-timeout header; this is the upstream Odoo timeout, not the local httpx timeout. Non-positive values are dropped (proxy default kicks in).

params / keyword shapes

params is wrapped to become positional args to Odoo's execute_kw(model, method, args, kwargs). The most common shapes:

# search / search_count / search_read: [[domain]]
session.search("res.partner", params=[[["is_company", "=", True]]])

# read: [ids, fields]
session.read("res.partner", params=[[1, 2, 3], ["name", "email"]])

# create: [vals]
session.create("res.partner", params=[{"name": "Acme"}])

# write: [ids, vals]
session.write("res.partner", params=[[1], {"name": "Acme Inc."}])

# unlink: [ids]
session.unlink("res.partner", params=[[1]])

# search_read with fields/limit/offset → use keyword
session.search_read(
    "res.partner",
    params=[[["is_company", "=", True]]],
    keyword={"fields": ["id", "name"], "limit": 100, "offset": 0},
)

# call_method: positional args of the target Odoo method go in params
session.call_method("account.move", "action_post", params=[[invoice_id]])

If a shape isn't here, it matches whatever execute_kw expects — the spec is Odoo's, not the proxy's.

Ops endpoints on the client itself

client.about()              -> BuildInfo(build: str, version: str)
client.license()            -> LicenseInfo(licensee: str, valid_until: str, is_valid: bool)
                               # valid_until is "YYYY-MM-DD" (string, no tz)
client.metrics()            -> str                # Prometheus text format
client.odoo_version(url)    -> JsonValue          # Odoo's version banner (dict)

---

v2 sessions — Odoo 19+ (JSON-2)

ODXProxy 0.9.0+ has /v2 endpoints that reach Odoo over its JSON-2 API instead of /jsonrpc. A v2 session is created from the same client:

from terrakernel.odxproxyclient import ODXProxyClient

with ODXProxyClient("https://proxy.example.com", "PROXY_X_API_KEY") as client:
    erp = client.for_instance_v2(
        url="https://erp.example.com",
        db="prod",
        api_key="ODOO_USER_API_KEY",     # an Odoo API key, not a password
        context={"lang": "en_US", "tz": "Asia/Jakarta", "allowed_company_ids": [1]},
    )

    partners = erp.search_read("res.partner", [["is_company", "=", True]],
                               fields=["name", "email"], limit=20, order="name asc")
    ids = erp.search("res.partner", [["customer_rank", ">", 0]], limit=10)    # list[int]
    n = erp.search_count("res.partner", [])                                   # int
    rows = erp.read("res.partner", ids, fields=["name"])                      # list[dict]
    schema = erp.fields_get("res.partner", attributes=["type", "string"])     # dict

    new_ids = erp.create("res.partner", [{"name": "Acme"}, {"name": "Globex"}])  # [41, 42]
    one = erp.create_one("res.partner", {"name": "Initech"})                    # 43
    erp.write("res.partner", new_ids, {"comment": "via v2"})                    # True
    erp.unlink("res.partner", new_ids)                                          # True

    erp.call_method("account.move", "action_post", ids=[7])
    erp.call_method("res.partner", "name_search", kwargs={"name": "Acm", "limit": 5})

The async client has the same methods on AsyncSessionV2: aclient.for_instance_v2(...), then await erp.search_read(...).

When to use which:

Odoo version	v1 Session	SessionV2
18 or older	✅ only option	❌ (Json2UnavailableError)
19 – 21	✅	✅
22 and newer	❌ (/jsonrpc removed)	✅ only option

client.supports_v2(odoo_url) checks once and caches the answer per URL. client.odoo_version_v2(odoo_url) returns {"version_info": [...], "version": "..."}.

How v2 differs from v1:

• Named arguments only. There are no params and no keyword. Each argument goes to Odoo under its Python parameter name: domain, fields, offset, limit, order, ids, vals, allfields, attributes. call_method takes ids= for record methods and kwargs={...} for everything else. Odoo rejects unknown names with OdooValidationError, and also rejects ids sent to @api.model methods.
• The domain is the list itself: [["is_company", "=", True]]. v1's extra wrapping list is gone.
• Arguments left None are omitted, so Odoo's own defaults apply.
• create always returns a list of ids, even for a single dict. Use create_one for an int.
• No user_id. Odoo derives the user from the API key. On Odoo 20+ the key's scope must be rpc (the default), and keys of non-admin users expire. client.session_v2(instance) also accepts a v1 OdooInstance and ignores its user_id.
• Context: the session context is merged into every call, and a call's context= keys win. Odoo applies no company selection unless allowed_company_ids is sent.
• Multi-database hosts: the database is selected by header and filtered by the server's dbfilter. If the host picks the database from its hostname, url must be that database's own hostname. Otherwise calls raise Json2UnavailableError.
• Binary fields on Odoo 20+ read as {"content": ..., "filename": ..., "size": ...} rather than a base64 string. This is an Odoo 20 change and applies to v1 too.

---

FastAPI integration

Create one AsyncODXProxyClient for the whole app, manage its lifetime with lifespan, and inject sessions via Depends. The single client owns the HTTP/2 connection pool — sharing it across requests is the whole performance story.

# app/main.py
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI, HTTPException, Request
from terrakernel.odxproxyclient import (
    AsyncODXProxyClient,
    AsyncSession,
    AuthError,
    ODXProxyError,
    OdooInstance,
    OdooLogicError,
)

ODOO = OdooInstance(
    url="https://erp.example.com",
    db="prod",
    user_id=2,
    api_key="ODOO_USER_API_KEY",
)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.odx = AsyncODXProxyClient(
        base_url="https://proxy.example.com",
        api_key="PROXY_X_API_KEY",
        default_timeout_secs=15.0,
    )
    try:
        yield
    finally:
        await app.state.odx.aclose()


app = FastAPI(lifespan=lifespan)


async def get_session(request: Request) -> AsyncSession:
    """Inject a session bound to the default Odoo instance."""
    return request.app.state.odx.session(ODOO)


@app.exception_handler(ODXProxyError)
async def odx_error_handler(_: Request, exc: ODXProxyError) -> HTTPException:
    # Translate proxy/Odoo failures into HTTP errors your API wants to expose.
    if isinstance(exc, AuthError):
        raise HTTPException(status_code=502, detail="upstream auth failed")
    if isinstance(exc, OdooLogicError):
        raise HTTPException(status_code=400, detail=exc.message)
    raise HTTPException(status_code=502, detail=exc.message)


@app.get("/partners")
async def list_partners(
    limit: int = 50,
    session: AsyncSession = Depends(get_session),
) -> list[dict]:
    rows = await session.search_read(
        "res.partner",
        params=[[]],
        keyword={"fields": ["id", "name", "email"], "limit": limit},
    )
    assert isinstance(rows, list)
    return rows


@app.post("/invoices/{invoice_id}/post")
async def post_invoice(
    invoice_id: int,
    session: AsyncSession = Depends(get_session),
) -> dict:
    await session.call_method("account.move", "action_post", params=[[invoice_id]])
    return {"ok": True}

Notes

• One AsyncODXProxyClient per process; never instantiate it per request.
• If you talk to multiple Odoo databases, keep one client and call client.session(other_instance) per request — Session/AsyncSession is cheap and stateless.
• For per-tenant credentials, build the OdooInstance from request state (auth token, header, etc.) inside the Depends instead of using a module constant.

---

Flask integration

The sync ODXProxyClient is thread-safe (it wraps a long-lived httpx.Client). Construct one at app-factory time and reuse it across requests; close it at process shutdown.

# app/__init__.py
import atexit

from flask import Flask, current_app, jsonify
from terrakernel.odxproxyclient import (
    ODXProxyClient,
    OdooInstance,
    OdooLogicError,
    ODXProxyError,
)

ODOO = OdooInstance(
    url="https://erp.example.com",
    db="prod",
    user_id=2,
    api_key="ODOO_USER_API_KEY",
)


def create_app() -> Flask:
    app = Flask(__name__)
    app.config.from_mapping(
        ODX_BASE_URL="https://proxy.example.com",
        ODX_API_KEY="PROXY_X_API_KEY",
    )

    odx = ODXProxyClient(
        base_url=app.config["ODX_BASE_URL"],
        api_key=app.config["ODX_API_KEY"],
        default_timeout_secs=15.0,
    )
    app.extensions["odx"] = odx
    atexit.register(odx.close)

    @app.errorhandler(OdooLogicError)
    def _odoo_logic(exc: OdooLogicError):
        return jsonify(error=exc.message, code=exc.code), 400

    @app.errorhandler(ODXProxyError)
    def _odx(exc: ODXProxyError):
        return jsonify(error=exc.message, code=exc.code), 502

    @app.get("/partners")
    def list_partners():
        session = current_app.extensions["odx"].session(ODOO)
        rows = session.search_read(
            "res.partner",
            params=[[]],
            keyword={"fields": ["id", "name", "email"], "limit": 50},
        )
        return jsonify(rows)

    @app.post("/invoices/<int:invoice_id>/post")
    def post_invoice(invoice_id: int):
        session = current_app.extensions["odx"].session(ODOO)
        session.call_method("account.move", "action_post", params=[[invoice_id]])
        return jsonify(ok=True)

    return app

Why atexit instead of teardown_appcontext? The client owns a connection pool meant to live for the whole process. teardown_appcontext fires after every request and would defeat the pool. Close at process exit.

If you're running Flask with a worker model (gunicorn, uwsgi), each worker process gets its own client — that's correct and what you want.

---

Error handling

Every wire-level failure becomes a typed exception, all inheriting from ODXProxyError. Catch the base if you want one handler; catch a subclass if you want to react specifically.

from terrakernel.odxproxyclient import (
    ODXProxyError,
    AuthError,             # -32000 / HTTP 401
    InvalidActionError,    # -32001 / HTTP 400
    MissingFnNameError,    # -32002 / HTTP 400
    OdooTimeoutError,      # -32003 / HTTP 504 (also: local httpx timeout)
    OdooConnectError,      # -32004 / HTTP 502 (also: local httpx connect error)
    InternalProxyError,    # -32005 / HTTP 500
    LicenseError,          # code 0 / HTTP 403
    OdooLogicError,        # any other code / HTTP 200 (Odoo pass-through)
    TransportError,        # unrecognized / malformed envelope
    # added with v2 (see "v2 sessions")
    Json2UnavailableError, # -32006 / HTTP 200: no JSON-2 (Odoo <= 18) or DB not selectable (dbfilter)
    InvalidRequestError,   # -32007 / HTTP 400: invalid model/method, or db/api_key not header-safe
)

try:
    session.write("res.partner", params=[[1], {"name": ""}])
except OdooLogicError as e:
    # Odoo-side validation error — 200 OK with an error envelope.
    print(e.code, e.message, e.data)
except OdooTimeoutError:
    # Either the proxy reported -32003 OR the local httpx call timed out.
    ...
except ODXProxyError as e:
    print(e.code, e.http_status, e.request_id)

Odoo's HTTP status as a subclass. On v2, every Odoo error carries Odoo's HTTP status as code. On v1 this only happens when Odoo itself answered non-2xx. These raise subclasses of OdooLogicError, so except OdooLogicError still catches them all:

Exception	code	Meaning	Retry?
OdooAuthError	401	Odoo rejected its API key: invalid, expired, wrong scope, or a password. This is not AuthError, which is the proxy key	no
OdooAccessError	403	Access rights, or a private (_-prefixed) method	no
OdooNotFoundError	404	Unknown model/method, or missing record	no
OdooConflictError	409	Odoo lock conflict	yes, with backoff
OdooValidationError	422	Validation error, or bad arguments	no
OdooServerError	5xx	Odoo server error	no

OdooLogicError.odoo_error_name returns Odoo's exception class from data["name"], e.g. "odoo.exceptions.ValidationError".

Code 0 is only LicenseError on HTTP 403. Odoo 19+ sends code 0 for every /jsonrpc error. Those arrive on HTTP 200 and raise OdooLogicError.

HTTP 200 is not always success. The proxy returns Odoo-side logic errors as HTTP 200 with an error object in the body. The client handles this for you — OdooLogicError will be raised. Do not implement your own status-code checks; let the typed exceptions do the work.

---

Performance notes

• One client per process, always. The connection pool, HTTP/2 multiplexing, and TLS session reuse all live on the httpx.Client / AsyncClient that the proxy client owns. Throwing it away per request throws those away too.
• HTTP/2 is on by default (http2=True). Disable with http2=False if your proxy speaks only HTTP/1.1.
• Response compression (gzip, br, deflate) is negotiated automatically by httpx — no SDK-side header juggling.
• orjson is used for both serialization and deserialization. Significantly faster than stdlib json for the kind of payloads search_read returns.
• No retries. This is deliberate. Wrap the client in your own retry policy (e.g. tenacity) if you need one — the SDK won't silently re-fire calls that might have side effects.

---

Advanced: shared httpx client

If you already manage a process-wide httpx.Client / AsyncClient and want the SDK to use it, pass it in. The SDK will not close clients it doesn't own.

import httpx
from terrakernel.odxproxyclient import AsyncODXProxyClient

http = httpx.AsyncClient(base_url="https://proxy.example.com", http2=True, timeout=15.0)
odx = AsyncODXProxyClient(
    base_url="https://proxy.example.com",
    api_key="PROXY_X_API_KEY",
    http_client=http,
)
# ... use odx ...
await odx.aclose()   # does NOT close `http`
await http.aclose()  # your responsibility

---

Testing

pip install -e ".[dev]"

# Unit tests (hermetic, respx-mocked, no network)
pytest -m "not integration"

# Integration tests against a real proxy
cp tests/integration/credentials.example.toml tests/integration/credentials.toml
$EDITOR tests/integration/credentials.toml
pytest -m integration

The v1 integration tests are read-only by design. They exercise about, license, metrics, odoo_version, search_count, search_read and fields_get, plus a negative auth case, and never create, write, or unlink. The integration suite auto-skips when credentials.toml is absent, so a fresh clone runs only the unit tests.

The v2 integration tests (test_integration_v2.py) are opt-in. Set [tests] v2 = true only when the proxy is ODXProxy 0.9.0+ and the Odoo is 19+. Their create/write/unlink lifecycle runs only with v2_writes = true, because it creates and deletes res.partner records.

---

License

MIT — see LICENSE. Copyright © 2026 Terrakernel.