REST quickstart

This example searches an existing library and prints a temporary link to a source file. You can also upload a small document and wait for it to become searchable. It does not create libraries, change permissions or delete data.

You need Python 3.9 or later, your TKP application URL and a token with API access and Read permission for a library. See API access to create the token. To get search results without uploading, the library must already contain processed content.

1. Set the application URL and token

Download quickstart.py and open a terminal in the folder where you saved it. If you use a repository checkout instead, run cd services/docs/examples/rest from its root. All commands below run from the folder containing quickstart.py.

Set the application URL and read the token into an environment variable:

export TKP_URL='https://tkp.example.com'
read -r -s TKP_TOKEN
export TKP_TOKEN
python3 quickstart.py

Replace the example URL with your TKP application URL. Use the origin only, without /api, /v1 or another path. Use HTTPS outside local development. After the read command, paste your token and press Enter; its characters are hidden.

The script verifies authentication and prints the workspace name followed by the libraries the token can read. Each line gives a library ID, name and effective permissions, for example:

Workspace: Example team
lib_REPLACE_ME: Manuals (read, resource.create)
Run again with --library ID and --query TEXT.

Copy a real library ID from your output for the next step. If no libraries appear, check both the token's permissions and its owner's library membership in TKP.

The underlying authentication check is:

curl --fail-with-body "$TKP_URL/api/v1/auth/check" \
  -H "Authorization: Bearer $TKP_TOKEN" \
  -H 'Accept: application/json'

2. Search an existing library

Choose an ID printed by the first run:

python3 quickstart.py \
  --library lib_REPLACE_ME \
  --query 'terms from one of your documents'

The example uses strategy=lexical, which matches terms in the content without calling an AI provider. It prints up to five candidates, including their source excerpts when available, and issues a five-minute source URL for the first candidate. Opening that URL lets you check the original file without sending your token to a browser. The URL itself grants temporary access, so do not write it to public logs.

In your own REST requests, omit strategy to use TKP's default balanced search. It also considers meaning rather than only matching terms. This can call the configured embedding service to represent the query and a reranking service to reorder the results by relevance.

To make the same library-scoped search directly with cURL, run:

curl --fail-with-body --get "$TKP_URL/api/v1/search/resources" \
  -H "Authorization: Bearer $TKP_TOKEN" \
  -H 'Accept: application/json' \
  --data-urlencode 'query=your search terms' \
  --data-urlencode 'library_ids[]=lib_REPLACE_ME' \
  --data-urlencode 'strategy=lexical' \
  --data-urlencode 'limit=5'

Use bracket notation for query arrays, for example repeated library_ids[] values. Without a library filter, search covers every library the token can read. Empty results means no candidates matched; it does not mean that authentication failed.

3. Upload and processing

This optional step adds a file to TKP. The token also needs Create resources in the selected library. Replace ./manual.pdf with a small PDF, Markdown file or another format supported by your instance:

python3 quickstart.py \
  --library lib_REPLACE_ME \
  --file ./manual.pdf \
  --query 'terms from the manual' \
  --wait-seconds 300

The script prints the resource ID, waits for processing to finish and then runs the search. TKP places the uploaded file in the library's .inbox folder.

Uploading triggers the instance's processing pipeline, including configured AI providers. This can consume provider credits. The example loads the file into memory; use a streaming multipart client for large files. Upload limits and available processing capabilities depend on the instance.

POST /api/v1/libraries/{library}/files accepts multipart field file. A new resource returns 201; a byte-identical duplicate in that library returns 200 with duplicate: true and the existing resource ID. Store that ID before waiting. A successful HTTP upload does not mean the resource is searchable.

To check progress, repeat GET /api/v1/libraries/{library}/resources/{resource} at intervals until processing finishes or your deadline is reached. This is called polling. Use the returned ingestion_status to decide whether to keep waiting:

ingestion_status	Client action
not_started, running	Wait and poll again within a deadline.
completed	Search the resource.
completed_with_warnings	Searchable, but some enrichment has reduced quality.
partial, failed	Stop polling and ask the operator to inspect the resource.

On timeout, keep the ID and inspect the resource in the admin panel. Do not re-upload automatically. A slow or unavailable worker can leave processing pending.

The executable example uses this polling loop:

def wait_until_searchable(client, library_id, resource_id, *, wait_seconds=300, interval=3):
    deadline = time.monotonic() + wait_seconds
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("Ingestion is still pending. Check the resource in the admin panel; do not re-upload automatically.")
        # Bound each request by the remaining polling window.
        original_timeout = client.timeout
        client.timeout = min(original_timeout, remaining)
        try:
            resource = client.request("GET", f"/libraries/{segment(library_id)}/resources/{segment(resource_id)}")
        finally:
            client.timeout = original_timeout
        state = resource["ingestion_status"]
        if state in ("completed", "completed_with_warnings"):
            return resource
        if state in ("partial", "failed"):
            raise ApiError(f"Ingestion ended with {state}. Ask the operator to inspect the resource.")
        if state not in ("not_started", "running"):
            raise ApiError(f"Unknown ingestion state: {state}")
        time.sleep(min(interval, max(0, deadline - time.monotonic())))

4. Show the evidence

The search output groups source excerpts under each matching resource. Use the resource ID printed in that output wherever an endpoint needs {resource}. Detailed search results contain results[].resource, including id, title, library, paths and relevant assets. An asset is an extracted piece of content, such as a text excerpt or image. Text assets can contain markdown and trace entries identifying source pages, regions or time ranges. Fields depend on the resource and processing output; do not assume every hit has text or a page. output_format=compact omits the abstract and assets.

Your backend can download the original file from GET /api/v1/resources/{resource} with the bearer token. To display it in a browser without exposing the token, request a temporary view URL instead. Replace res_REPLACE_ME with a resource ID from your search results:

curl --fail-with-body "$TKP_URL/api/v1/resources/res_REPLACE_ME/view-urls" \
  -H "Authorization: Bearer $TKP_TOKEN" \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json' \
  --data '{"ttl_minutes":5,"disposition":"inline"}'

The JSON response has url and expires_in_minutes. Use the returned URL unchanged; modifying signed query parameters invalidates it. Page, page-region and asset view-URL endpoints can show the exact evidence instead of the entire original. Use the API reference for their required parameters.

Complete runnable example

This is the source of the downloadable script used above. You can also save it as quickstart.py and run it with the same environment variables and arguments.

Complete source

#!/usr/bin/env python3
"""Authenticate, optionally upload, search, and issue a temporary source URL."""

import argparse
import json
import mimetypes
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path


class ApiError(RuntimeError):
    pass


class NoRedirect(urllib.request.HTTPRedirectHandler):
    # Do not forward credentials to a redirected host.
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


# region client
class Client:
    def __init__(self, base_url, token, timeout=30):
        parsed = urllib.parse.urlsplit(base_url)
        if (parsed.scheme not in ("http", "https") or not parsed.netloc
                or parsed.path not in ("", "/") or parsed.query or parsed.fragment
                or parsed.username or parsed.password):
            raise ValueError("TKP_URL must be an origin, for example https://tkp.example.com")
        self.base_url = base_url.rstrip("/") + "/api/v1"
        self.token = token
        self.timeout = timeout
        self.opener = urllib.request.build_opener(NoRedirect())

    def request(self, method, path, *, query=None, payload=None, body=None, content_type=None):
        url = self.base_url + path
        if query:
            url += "?" + urllib.parse.urlencode(query, doseq=True)
        headers = {"Authorization": f"Bearer {self.token}", "Accept": "application/json"}
        if payload is not None:
            body = json.dumps(payload).encode()
            content_type = "application/json"
        if content_type:
            headers["Content-Type"] = content_type
        request = urllib.request.Request(url, data=body, headers=headers, method=method)
        try:
            with self.opener.open(request, timeout=self.timeout) as response:
                return json.load(response)
        except urllib.error.HTTPError as error:
            raw = error.read()
            try:
                problem = json.loads(raw)
                detail = problem.get("detail", problem.get("message", error.reason))
                fields = problem.get("errors")
                if fields:
                    detail += " " + json.dumps(fields)
            except (ValueError, AttributeError):
                detail = str(error.reason)
            raise ApiError(f"HTTP {error.code}: {detail}") from None

    def libraries(self):
        cursor = None
        seen = set()
        while True:
            page = self.request("GET", "/libraries", query={"limit": 50, **({"cursor": cursor} if cursor else {})})
            yield from page["items"]
            cursor = page["next_cursor"]
            if cursor is None:
                return
            if cursor in seen:
                raise ApiError("The server returned a repeated pagination cursor")
            seen.add(cursor)

    def upload(self, library_id, path):
        boundary = "tkp-" + uuid.uuid4().hex
        filename = path.name.replace('"', "_").replace("\r", "_").replace("\n", "_")
        mime = mimetypes.guess_type(filename)[0] or "application/octet-stream"
        body = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n'
                f"Content-Type: {mime}\r\n\r\n").encode() + path.read_bytes() + f"\r\n--{boundary}--\r\n".encode()
        return self.request("POST", f"/libraries/{segment(library_id)}/files", body=body,
                            content_type=f"multipart/form-data; boundary={boundary}")
# endregion client


def segment(value):
    return urllib.parse.quote(str(value), safe="")


# region polling
def wait_until_searchable(client, library_id, resource_id, *, wait_seconds=300, interval=3):
    deadline = time.monotonic() + wait_seconds
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("Ingestion is still pending. Check the resource in the admin panel; do not re-upload automatically.")
        # Bound each request by the remaining polling window.
        original_timeout = client.timeout
        client.timeout = min(original_timeout, remaining)
        try:
            resource = client.request("GET", f"/libraries/{segment(library_id)}/resources/{segment(resource_id)}")
        finally:
            client.timeout = original_timeout
        state = resource["ingestion_status"]
        if state in ("completed", "completed_with_warnings"):
            return resource
        if state in ("partial", "failed"):
            raise ApiError(f"Ingestion ended with {state}. Ask the operator to inspect the resource.")
        if state not in ("not_started", "running"):
            raise ApiError(f"Unknown ingestion state: {state}")
        time.sleep(min(interval, max(0, deadline - time.monotonic())))
# endregion polling


# region workflow
def run(client, args):
    context = client.request("GET", "/auth/check")
    print(f"Workspace: {context['tenant']['name']}")
    libraries = list(client.libraries())
    for library in libraries:
        print(f"{library['id']}: {library['name']} ({', '.join(library['access'])})")
    if not args.library:
        print("Run again with --library ID and --query TEXT.")
        return
    selected = next((item for item in libraries if item["id"] == args.library), None)
    if selected is None or "read" not in selected["access"]:
        raise ApiError("Choose a listed library with read access")
    if args.file:
        if "resource.create" not in selected["access"]:
            raise ApiError("Uploading also requires resource.create access")
        uploaded = client.upload(args.library, args.file)
        print(f"Resource: {uploaded['id']}; duplicate: {uploaded.get('duplicate', False)}")
        ready = wait_until_searchable(client, args.library, uploaded["id"], wait_seconds=args.wait_seconds)
        print(f"Ingestion: {ready['ingestion_status']}")
    result = client.request("GET", "/search/resources", query={
        "query": args.query, "library_ids[]": [args.library], "strategy": "lexical", "limit": 5,
    })
    print(result["notice"])
    for hit in result["results"]:
        resource = hit["resource"]
        print(json.dumps(resource, ensure_ascii=False, indent=2))
    if result["results"]:
        resource_id = result["results"][0]["resource"]["id"]
        view = client.request("POST", f"/resources/{segment(resource_id)}/view-urls",
                              payload={"ttl_minutes": 5, "disposition": "inline"})
        print(f"First candidate source (expires in {view['expires_in_minutes']} minutes): {view['url']}")
    else:
        print("No candidates. Try terms from a processed document or a different permitted library.")
# endregion workflow


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--library", help="Opaque library ID from the first run")
    parser.add_argument("--query", help="Lexical search terms")
    parser.add_argument("--file", type=Path, help="Optional small supported file to upload (read into memory)")
    parser.add_argument("--wait-seconds", type=float, default=300, help="Maximum ingestion wait (default: 300)")
    args = parser.parse_args()
    if args.library and not args.query:
        parser.error("--library requires --query")
    if args.file and not args.library:
        parser.error("--file requires --library and --query")
    if args.wait_seconds <= 0:
        parser.error("--wait-seconds must be positive")
    token = os.environ.get("TKP_TOKEN")
    if not token or not os.environ.get("TKP_URL"):
        parser.error("Set TKP_URL (origin) and TKP_TOKEN")
    try:
        run(Client(os.environ["TKP_URL"], token), args)
    except (ApiError, ValueError, KeyError, OSError, TimeoutError, urllib.error.URLError) as error:
        print(f"Error: {error}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

To run the script's contract tests, use a repository checkout and run this command from its root. The tests use a local HTTP fixture, without TKP or AI providers:

python3 -m unittest discover -s services/docs/examples/rest -v