Python Sample — Context API

Full workflow: authenticate → get presigned URL → upload file → submit enrichment → poll results.

Download the Python Notebook (.ipynb)

---

Setup

import base64
import json
import time
from urllib.parse import urlencode, quote
from urllib.request import Request, urlopen

CONTEXT_API_URL = "https://knowledge-enrichment.ai.app.hyland.com/latest/api/context-enrichment"  # replace with your environment URL
OAUTH_URL = "https://auth.app.hyland.com/idp"
CLIENT_ID = "YOUR_CLIENT_ID"
CLIENT_SECRET = "YOUR_CLIENT_SECRET"

---

Workflow

1. Get access token

def get_access_token():
    headers = {
        "Accept": "application/json",
        "Authorization": f"Basic {base64.b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()}"
    }
    body = urlencode({
        "grant_type": "client_credentials",
        "scope": "environment_authorization"
    }).encode()
    req = Request(f"{OAUTH_URL}/connect/token", data=body, headers=headers, method="POST")
    with urlopen(req) as resp:
        return json.loads(resp.read())["access_token"]

token = get_access_token()

2. Get presigned upload URL

def get_presigned_url(token, content_type):
    encoded_type = quote(content_type, safe="")
    req = Request(
        f"{CONTEXT_API_URL}/files/upload/presigned-url?contentType={encoded_type}",
        headers={"Authorization": f"Bearer {token}"},
        method="GET"
    )
    with urlopen(req) as resp:
        return json.loads(resp.read())

presign = get_presigned_url(token, "application/pdf")
presigned_url = presign["presignedUrl"]
object_key = presign["objectKey"]
print(f"objectKey: {object_key}")

3. Upload file

def upload_file(presigned_url, file_path, content_type):
    with open(file_path, "rb") as f:
        data = f.read()
    req = Request(
        presigned_url,
        data=data,
        headers={"Content-Type": content_type},
        method="PUT"
    )
    urlopen(req)
    print("Upload complete")

upload_file(presigned_url, "input/document.pdf", "application/pdf")

4. Submit enrichment job

def process(token, object_key, actions):
    body = json.dumps({
        "version": "context.api/v2",
        "objectKeys": [{"path": object_key}],
        "actions": actions
    }).encode()
    req = Request(
        f"{CONTEXT_API_URL}/content/process",
        data=body,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        },
        method="POST"
    )
    with urlopen(req) as resp:
        return json.loads(resp.read())

result = process(token, object_key, {
    "textSummarization": {
        "maxWordCount": 150
    },
    "textClassification": {
        "classes": ["Report", "Contract", "Invoice", "Other"]
    }
})
processing_id = result["processingId"]
print(f"Processing ID: {processing_id}")

5. Poll results

def poll_results(token, processing_id, interval=3, timeout=300):
    elapsed = 0
    while elapsed < timeout:
        req = Request(
            f"{CONTEXT_API_URL}/content/process/{processing_id}/results",
            headers={"Authorization": f"Bearer {token}"},
            method="GET"
        )
        try:
            with urlopen(req) as resp:
                if resp.status == 202:
                    # Still processing - 202 Accepted
                    data = json.loads(resp.read())
                    print(f"Status: {data.get('status', 'PROCESSING')}, inProgress: {data.get('inProgress', True)}")
                else:
                    # Complete - 200 OK
                    data = json.loads(resp.read())
                    return data
        except Exception as e:
            print(f"Error polling results: {e}")
            raise
        time.sleep(interval)
        elapsed += interval
    raise TimeoutError(f"Job {processing_id} did not complete within {timeout}s")

results = poll_results(token, processing_id)
print(json.dumps(results, indent=2))