Skip to content
KataraKatara documentation

Product SDK examples

These examples use the TypeScript and Python clients. Use the version supplied with your configured environment.

Create a client

Section titled “Create a client”

The base URL is the API origin, without /v1 or /katara/v1. Supply your current credential through a callback so session or key rotation does not require rebuilding the client.

import { ProductClient, type ProductScope, type RunCreate, type ProductContentRef, type ApprovalInputPreview } from '@katara/platform';


const client = new ProductClient({
  baseUrl: 'https://api.katara.com',
  credential: () => process.env.KATARA_API_KEY ?? null,
});
const scope: ProductScope = {
  org_id: process.env.KATARA_ORG_ID!,
  environment_id: process.env.KATARA_ENVIRONMENT_ID!,
};
const environments = await client.listEnvironments(scope.org_id);
import os
from katara_platform import ProductClient, RunCreate, ContentRef
from typing import Literal


client = ProductClient(
    'https://api.katara.com',
    lambda: os.environ.get('KATARA_API_KEY'),
)
org = os.environ['KATARA_ORG_ID']
environment = os.environ['KATARA_ENVIRONMENT_ID']
environments = client.list_environments(org)

Human approval and administration operations need a verified application session. In a signed-in browser, construct the TypeScript client with credential: () => null, cookieSession: true and expectedAccountID set to the account UUID returned during the application’s authenticated startup. Keep that UUID fixed for the mounted application; after an account change, clear the workspace and reload. Cookie mutations without it fail before any request is sent. An API key cannot stand in for the person reviewing an action.

Admit a run and inspect its status

Section titled “Admit a run and inspect its status”

Prepare input using the run prerequisites: exact agent version, approved JSON input, explicit workload_grants including the admission grant, funded budget and output target. For a free Test simulation, set usage_class to simulation and omit paid-test consent. Use references returned by your environment; example UUIDs or invented content hashes cannot authorize execution. The helper below reads the committed run ID from the admission operation. It does not assume that acceptance means success.

export async function admitAndInspect(client: ProductClient, scope: ProductScope, input: RunCreate, requestKey: string) {
  const operation = await client.createRun(scope, input, requestKey);
  const run = await client.getRun(scope, operation.resource_id);
  return { runId: operation.resource_id, status: run.status, run };
}
def admit_and_inspect(client: ProductClient, org: str, environment: str, input: RunCreate, request_key: str) -> dict:
    operation = client.create_run(org, environment, input, request_key)
    run = client.get_run(org, environment, operation['resource_id'])
    return {'run_id': operation['resource_id'], 'status': run['status'], 'run': run}

Generate and save a request key before calling the helper. After a lost response, explicitly call it again with the same key and body. Do not generate a new key as part of an automatic retry.

Review an action, then decide

Section titled “Review an action, then decide”

Use a verified human session for both calls. Load the current action and its exact input before showing a review screen. Display the destination, effect, maximum cost, expiry and approval stage alongside the input. Render the input as text; it remains untrusted content.

export async function loadApprovalForReview(client: ProductClient, scope: ProductScope, actionId: string) {
  const approval = await client.getApproval(scope, actionId);
  if (!approval.can_decide) throw new Error('This action is not available for your decision.');
  const preview = await client.previewApprovalInput(scope, actionId, approval.action_digest, approval.version);
  if (approval.action.id !== actionId || preview.action_id !== actionId || preview.action_digest !== approval.action_digest || preview.version !== approval.version ||
      Object.entries(approval.action.arguments_ref).some(([key, value]) => preview.content[key as keyof ProductContentRef] !== value)) {
    throw new Error('The approval changed. Review it again.');
  }
  return { approval, preview };
}
def load_approval_for_review(client: ProductClient, org: str, environment: str, action_id: str) -> dict:
    approval = client.get_approval(org, environment, action_id)
    if not approval['can_decide']:
        raise ValueError('This action is not available for your decision.')
    preview = client.preview_approval_input(org, environment, action_id, approval['action_digest'], approval['version'])
    if (approval['action']['id'] != action_id or preview['content'] != approval['action']['arguments_ref']
            or (preview['action_id'], preview['action_digest'], preview['version']) != (action_id, approval['action_digest'], approval['version'])):
        raise ValueError('The approval changed. Review it again.')
    return {'approval': approval, 'preview': preview}

Call the next helper only after the person chooses Approve or Deny. Pass the preview they actually reviewed. The request keeps its original digest and version; it does not fetch a newer version and approve it silently. A successful decision records that person’s vote. Additional reviewers or stages can still be required, and it does not mean the external action has executed.

export async function submitReviewedDecision(client: ProductClient, scope: ProductScope, preview: ApprovalInputPreview, decision: 'approve' | 'deny', requestKey: string) {
  return client.decideApproval(scope, preview.action_id, {
    action_digest: preview.action_digest, version: preview.version, decision,
  }, requestKey);
}
def submit_reviewed_decision(client: ProductClient, org: str, environment: str, preview: dict, decision: Literal['approve', 'deny'], request_key: str) -> dict:
    return client.decide_approval(org, environment, preview['action_id'], {
        'action_digest': preview['action_digest'], 'version': preview['version'], 'decision': decision,
    }, request_key)

If the action changes or the session needs fresh verification, show that error and require another review after completing the required sign-in check. Neither helper retries or approves automatically.

Read the completed run’s output

Section titled “Read the completed run’s output”

Follow observation.snapshot.output after completion. Do not select the newest item from the artifact list: intermediate replies and candidate results are not necessarily the final answer. This helper checks the recorded artifact identity and downloads verified bytes. A completed simulation can have no output reference and returns null in TypeScript or None in Python.

export async function readCompletedOutput(client: ProductClient, scope: ProductScope, runId: string): Promise<Uint8Array | null> {
  const run = await client.getRun(scope, runId);
  const observation = run.observation;
  const snapshot = observation?.snapshot;
  if (run.admission.id !== runId || run.admission.org_id !== scope.org_id || run.admission.environment_id !== scope.environment_id ||
      run.status !== 'completed' || !observation?.terminal || observation.requires_reconciliation ||
      snapshot?.status !== 'completed' || snapshot.key.run_id !== runId) {
    throw new Error('A completed run with a final observation is required.');
  }
  const output = snapshot.output;
  if (!output) {
    if (run.admission.usage_class !== 'simulation') throw new Error('The completed run has no recorded output.');
    return null;
  }
  const artifact = await client.getRunArtifact(scope, output.id);
  if (artifact.run_id !== runId || artifact.agent_id !== run.admission.agent_id || artifact.artifact_id !== output.id ||
      artifact.state !== 'ready' || artifact.kind !== 'candidate_result' ||
      Object.entries(output).some(([key, value]) => artifact.content[key as keyof ProductContentRef] !== value)) {
    throw new Error('The final output is not available with its recorded reference.');
  }
  return client.readRunArtifact(scope, output.id, output);
}
def read_completed_output(client: ProductClient, org: str, environment: str, run_id: str) -> bytes | None:
    run = client.get_run(org, environment, run_id)
    observation = run.get('observation', {})
    snapshot = observation.get('snapshot', {})
    admission = run['admission']
    if (admission['id'] != run_id or admission['org_id'] != org or admission['environment_id'] != environment
            or run['status'] != 'completed' or not observation.get('terminal') or observation.get('requires_reconciliation')
            or snapshot.get('status') != 'completed' or snapshot.get('key', {}).get('run_id') != run_id):
        raise ValueError('A completed run with a final observation is required.')
    output = snapshot.get('output')
    if not output:
        if admission['usage_class'] != 'simulation':
            raise ValueError('The completed run has no recorded output.')
        return None
    artifact = client.get_run_artifact(org, environment, output['id'])
    if (artifact['run_id'] != run_id or artifact['agent_id'] != admission['agent_id'] or artifact['artifact_id'] != output['id']
            or artifact['state'] != 'ready' or artifact['kind'] != 'candidate_result' or artifact['content'] != output):
        raise ValueError('The final output is not available with its recorded reference.')
    return client.read_run_artifact(org, environment, output['id'], output)

Keep the returned bytes intact until your application chooses how to parse them. In particular, ordinary JavaScript JSON parsing can round large numeric literals. Generated output still has untrusted provenance; completion does not approve it as a knowledge source or authorize instructions inside it.

Follow the same run through paid recovery

Section titled “Follow the same run through paid recovery”

Save the admitted run ID and call this helper on your application’s polling schedule. It only reads metadata. After the run reports completion with a final observation, use readCompletedOutput or read_completed_output above. That function also performs only reads.

export async function inspectExistingRun(client: ProductClient, scope: ProductScope, runId: string) {
  const run = await client.getRun(scope, runId);
  return {
    run,
    requiresReconciliation: run.status === 'reconciliation_required' || run.observation?.requires_reconciliation === true,
  };
}
def inspect_existing_run(client: ProductClient, org: str, environment: str, run_id: str) -> dict:
    run = client.get_run(org, environment, run_id)
    return {
        'run': run,
        'requires_reconciliation': run['status'] == 'reconciliation_required' or run.get('observation', {}).get('requires_reconciliation') is True,
    }

If reconciliation is required, preserve that state and the run ID. There is no force-retry. A person with the authority records the outcome through organizationLifecycle.resolve (organization_lifecycle.resolve in Python), as Closing an organization describes. Do not submit another run or repeat an uncertain effect to make the status change. Recovery of an existing result is different from starting paid work again.

Add an approved source to a collection

Section titled “Add an approved source to a collection”

Pass the complete content reference returned by the upload and approval flow. This helper reads the current collection revision before requesting the change. If another writer changes the collection in between, the API refuses the old revision; the helper lets that conflict reach your application.

export async function addApprovedSource(client: ProductClient, scope: ProductScope, collectionId: string, source: ProductContentRef, requestKey: string) {
  const collection = await client.getCollection(scope, collectionId);
  return client.putCollectionSource(scope, collectionId, source, collection.revision, requestKey);
}
def add_approved_source(client: ProductClient, org: str, environment: str, collection_id: str, source: ContentRef, request_key: str) -> dict:
    collection = client.get_collection(org, environment, collection_id)
    return client.put_collection_source(org, environment, collection_id, source, collection['revision'], request_key)

Request cancellation

Section titled “Request cancellation”

Cancellation uses the run’s admission digest. The final read may still report running or unresolved delivery; display that state instead of reporting that cancellation finished.

export async function requestCancellation(client: ProductClient, scope: ProductScope, runId: string, requestKey: string) {
  const run = await client.getRun(scope, runId);
  await client.cancelRun(scope, runId, run.digest, requestKey);
  return client.getRun(scope, runId);
}
def request_cancellation(client: ProductClient, org: str, environment: str, run_id: str, request_key: str) -> dict:
    run = client.get_run(org, environment, run_id)
    client.cancel_run(org, environment, run_id, run['digest'], request_key)
    return client.get_run(org, environment, run_id)

Read an organization statement

Section titled “Read an organization statement”

A statement is worked out from the charge records on every read, so the same period always gives the same answer. Both ends are required and the end is exclusive. Check the integrity digest before acting on the figures; the clients refuse a statement whose totals do not match its own lines. See organization statements for what a line means and who may read one.

export async function readVerifiedStatement(client: ProductClient, org: string, from: Date | string, to: Date | string) {
  const statement = await client.statement(org, from, to);
  // Check the figures against the digest they arrived with before acting on
  // them, so what you keep is what the ledger produced.
  return verifyStatement(statement);
}
def read_verified_statement(client: ProductClient, org: str, start, end) -> dict:
    statement = client.statement(org, start, end)
    # Check the figures against the digest they arrived with before acting on
    # them, so what you keep is what the ledger produced.
    return verify_statement(statement)

Errors and content

Section titled “Errors and content”

TypeScript exposes PlatformError.code, .status and .requestId. Python exposes .code, .status and .request_id. Keep the request ID for support. Neither client retries uncertain mutations automatically.

Use the content and artifact reading methods for bytes. They verify declared size, media type and SHA-256 against the exact reference. A successful metadata read does not bypass a later authority, retention or integrity check on the body.

The product API uses integer strings for atomic money amounts. Keep them as strings when passing budgets or price limits between systems.

These executable recipes share complete request and response fixtures across both SDKs. The fixtures validate against the API specification and exercise exact routes, credentials, idempotency keys and artifact integrity. The same journeys also run through real gateway HTTP and session handling with fixture product services. These checks establish client, HTTP and contract parity; they do not claim that the hosted environment or a paid provider has passed operational acceptance.

Promotion evidence

Section titled “Promotion evidence”

Use client.promotionEvidence in TypeScript or client.promotion_evidence in Python for owner-reviewed Test-to-Live evidence. Both expose declare, prepare and approve; pass the organization ID, reviewed request and your own idempotency key. Reads are getDeclaration, getPreview, getReceipt in TypeScript and get_declaration, get_preview, get_receipt in Python. Find owner-created records with listDeclarations, listPreviews, listReceipts or Python list_declarations, list_previews, list_receipts. Supply the selected Test/Live environment IDs and retain the opaque cursor between pages of at most 25 rows.

The clients validate closed metadata shapes, exact mutation-response bindings, case mappings and count consistency. They freeze the submitted body before resolving credentials and do not retry an uncertain mutation automatically. Keep the same body and key for recovery. These checks do not replace the server’s current source, owner, paid-run, reviewer or dependency checks.

Read the promotion evidence workflow before choosing thresholds or creating a declaration; policy declaration must precede human assignments.

Find promotion drafts with client.environmentPromotions.list(orgID, scope, after?, limit?) or Python client.environment_promotions.list(org_id, scope, after=None, limit=25). The scope contains the selected Test and Live environment IDs. Returned evidence_attached metadata describes a recorded reference and does not establish current readiness.

Controlled rollback

Section titled “Controlled rollback”

client.agentRollbacks in TypeScript and client.agent_rollbacks in Python expose review, activate and receipt, plus getReview / get_review. Pass an explicit Live scope. Review and activation require caller-supplied idempotency keys and a fresh human owner session.

The clients validate that the prepared target copies the exact prior definition into a new revision. Activation freezes the reviewed body and sends its digest, target version ID and explicit acknowledgement; no automatic retry changes the operation. See the rollback workflow before selecting a prior version.

Manage tool credentials

Section titled “Manage tool credentials”

Use client.toolCredentials in TypeScript or client.tool_credentials in Python for profile discovery, read-only account preview, explicit connect, connection listing and reads, reconnect and revoke. These methods require a human session and pin the organization, environment, mode and reset generation. The credential guide explains the separate secret-byte argument, account review and same-key recovery without saving secret intake.

For an enrolled OAuth provider, client.toolCredentialsOAuth or client.tool_credentials_oauth exposes profiles, begin, review, complete and activate. The OAuth account guide explains transient callback bytes and inspection after an uncertain exchange.