Python Client

clinia-context-engine is the client for Python 3.10 or later. It comes as a synchronous client and an asynchronous client with the same methods.

Install and authenticate

Every call to a workspace needs an access token. The client gets that token for you once you give it the client ID and secret of a credential.

You need three things before you start:

• A workspace and its URL, https://<workspace-id>.w.clinia.cloud. See Organizations, Workspaces and Environments.
• An OAuth client credential for that workspace, created in the Console. See Authentication and Credentials.
• Python 3.10 or later.

pip install clinia-context-engine

from clinia_context_engine import ClientCredentials, ContextEngineClient

with ContextEngineClient(
    base_url="CLINIA_WORKSPACE_URL",
    auth=ClientCredentials(
        client_id="CLINIA_CLIENT_ID",
        client_secret="CLINIA_CLIENT_SECRET",
    ),
) as client:
    ...  # The call in the next step goes here.

    info = client.info.get_service_info()
    print(info.name, info.status)
    # Clinia Context Engine healthy

The client requests an access token on the first call and requests a new one a minute before that token expires. Calls made while a token request is in flight wait for that same request.

Use the async client

For async code, AsyncContextEngineClient takes the same options and has the same methods. Create it with async with and await each call.

import asyncio

from clinia_context_engine import AsyncContextEngineClient, ClientCredentials


async def main() -> None:
    async with AsyncContextEngineClient(
        base_url="CLINIA_WORKSPACE_URL",
        auth=ClientCredentials(
            client_id="CLINIA_CLIENT_ID",
            client_secret="CLINIA_CLIENT_SECRET",
        ),
    ) as client:
        info = await client.info.get_service_info()
        print(info.name, info.status)


asyncio.run(main())

Share one token across clients

A client keeps its token in its own memory, so every client you create requests a token of its own. To let several clients use the same token, create one token provider and pass it as auth to each client. The provider caches the token and renews it a minute before it expires.

Create the provider with ClientCredentialsTokenProvider, or AsyncClientCredentialsTokenProvider for the async client.

from clinia_context_engine import (
    ClientCredentials,
    ClientCredentialsTokenProvider,
    ContextEngineClient,
)

base_url = "CLINIA_WORKSPACE_URL"
get_access_token = ClientCredentialsTokenProvider(
    ClientCredentials(
        client_id="CLINIA_CLIENT_ID",
        client_secret="CLINIA_CLIENT_SECRET",
    )
)

ingest_client = ContextEngineClient(base_url=base_url, auth=get_access_token)
read_client = ContextEngineClient(base_url=base_url, auth=get_access_token)

A provider is a function that returns a token, and the client calls it before every request. To keep the token in a cache that several processes share, or to decide yourself when it is renewed, pass auth your own function instead. It should return the cached token and request a new one only when the cached token is about to expire.

def get_cached_access_token() -> str:
    """Returns the cached access token, and requests a new one when it is about to expire."""
    ...


client = ContextEngineClient(base_url=base_url, auth=get_cached_access_token)

Test against a local stub

To run tests against a local stub or recorded responses, create the client without auth. It then sends no Authorization header and accepts an http:// URL.

client = ContextEngineClient(base_url="http://localhost:8000")

Call the API

Methods are grouped by resource on the client: info, patients, ingest, sessions, vfs and graph. The REST reference page of a method's operation documents its parameters, body and response.

Arguments

A method takes its arguments in this order:

• Path parameters come first, as positional arguments in the order they appear in the path, such as the patient ID and then the node type for graph.list_nodes.
• The request body comes next, as one argument, either the model or a plain dict.
• Query parameters come last, as keyword arguments such as page=0, per_page=50.

Methods

Every method has a page in the reference, under its resource, with the call and a link to the operation's REST reference page, which documents its parameters. For example, Get patient is patients.get_patient.

Types

Each schema of the API is a Pydantic model in models, imported with from clinia_context_engine import models.

from clinia_context_engine import models


def describe_patient(patient: models.Patient) -> str:
    return f"{patient.id}, registered {patient.created_at}"

Responses and errors

A call can fail in three ways: the API answers with an error status, the request gets no response, or the client cannot get an access token.

Failures

Every exception the client raises inherits from ContextEngineError.

Handle a failed call

A method returns the parsed response body and raises one of the exceptions above when the call fails. Wrap a call in try and catch APIStatusError to read the status and the error type.

from clinia_context_engine import APIStatusError

try:
    patient = client.patients.get_patient("patient-123")
    print(patient.id, patient.name)
except APIStatusError as err:
    print(err.status_code, err.error.type if err.error else err.body)

The error body

The client parses the error body into the ApiError type, whose fields and type values are documented in the API reference.

Customize the HTTP layer

The http_client option, which takes an httpx.Client, gives you control over how the client sends its requests, for logging, tracing, a proxy, a timeout or mocked responses in tests.

Log requests and add headers

Create an httpx.Client with event_hooks for request and response, and pass it to the client as http_client. The hooks then run on every request the client sends. The httpx package is installed with the client.

import httpx

from clinia_context_engine import ClientCredentials, ContextEngineClient


def log_response(response: httpx.Response) -> None:
    request = response.request
    print(request.method, request.url, response.status_code)


with httpx.Client(event_hooks={"response": [log_response]}, timeout=60.0) as http_client:
    with ContextEngineClient(
        base_url="CLINIA_WORKSPACE_URL",
        auth=ClientCredentials(
            client_id="CLINIA_CLIENT_ID",
            client_secret="CLINIA_CLIENT_SECRET",
        ),
        http_client=http_client,
    ) as client:
        client.info.get_service_info()

Send requests through a proxy

Pass an httpx.Client configured with the proxy both as http_client and to a ClientCredentialsTokenProvider used as auth, so the token request goes through it too.

import httpx

from clinia_context_engine import (
    ClientCredentials,
    ClientCredentialsTokenProvider,
    ContextEngineClient,
)

credentials = ClientCredentials(
    client_id="CLINIA_CLIENT_ID",
    client_secret="CLINIA_CLIENT_SECRET",
)

with httpx.Client(proxy="http://proxy.example.com:3128", timeout=60.0) as http_client:
    get_access_token = ClientCredentialsTokenProvider(credentials, http_client=http_client)
    with ContextEngineClient(
        base_url="CLINIA_WORKSPACE_URL",
        auth=get_access_token,
        http_client=http_client,
    ) as client:
        client.info.get_service_info()

Mock the API in tests

Pass http_client an httpx.Client built on an httpx.MockTransport.

import httpx

from clinia_context_engine import ContextEngineClient


def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(
        200,
        json={
            "name": "Clinia Context Engine",
            "status": "healthy",
            "version": {"number": "1.0.0", "buildSha": "unknown", "buildDate": "unknown"},
        },
    )


with httpx.Client(transport=httpx.MockTransport(handler)) as http_client:
    client = ContextEngineClient(base_url="http://localhost:8000", http_client=http_client)

A mocked client usually runs without auth, as described in Test against a local stub, so that it makes no token request.

Set a timeout

The timeout option sets the time allowed for each request, in seconds or as an httpx.Timeout, and defaults to 60 seconds. It applies only when the client creates its own HTTP client, so set the timeout on the httpx.Client you pass as http_client otherwise.

from clinia_context_engine import ClientCredentials, ContextEngineClient

client = ContextEngineClient(
    base_url="CLINIA_WORKSPACE_URL",
    auth=ClientCredentials(
        client_id="CLINIA_CLIENT_ID",
        client_secret="CLINIA_CLIENT_SECRET",
    ),
    timeout=30.0,
)

Gotchas

• A client created without auth starts without error. Every call it then makes to a workspace returns a 401.
• The client keeps its connections open until it is closed, so create it in a with block or call close() when you are done.
• The client never closes an http_client you pass in, so close it yourself when you are done.
• The token request does not go through the http_client you pass, so a proxy or a mock transport set there does not reach it. Create a ClientCredentialsTokenProvider with that same http_client and pass it as auth to route both.