Python SDK

Use Algenta from Python — install algenta-sdk, authenticate, and call the engine with a typed client. Includes runnable examples.

This page is the reference for the live Python package Algenta ships today: algenta-sdk (imported as decision_engine), a thin typed HTTP client for the Algenta API. Use it for direct, typed access to hosted or self-hosted Algenta.

algenta-sdk is the live PyPI package. It needs no Mojo, worker, or runtime-service setup: the client connects to an Algenta deployment whose compute runtime is already integrated.

Install

Install the published HTTP client from PyPI.

# Typed HTTP client. Imported as `decision_engine`.
pip install algenta-sdk

The package requires Python 3.12 or newer. Pull-request CI builds the wheel and verifies a fresh install of it on Python 3.12, 3.13 and 3.14. It depends on httpx>=0.28 and pydantic>=2.7.

The import name is decision_engine, not algenta_sdk. This historical module name is stable.

Authentication and configuration

The client reads its API key, in order, from the api_key= argument, then ALGENTA_API_KEY, then the legacy DE_API_KEY. If none are set, the constructor raises ValueError.

1

Set your API key

Export a live or sandbox key into the environment. Live keys carry the de_live_ prefix; sandbox keys carry de_test_.

export ALGENTA_API_KEY="de_live_..."        # production key
# or, for the sandbox:
export ALGENTA_API_KEY="de_test_..."

2

Construct the client

Point base_url at the hosted API (the default) or at your own engine.

import os
from decision_engine import AlgentaClient

client = AlgentaClient(
    api_key=os.environ["ALGENTA_API_KEY"],
    base_url="https://api.algenta.ai",   # default; override for self-hosted
)

3

Verify the connection

Read the live contract version as a cheap round-trip. The client is a context manager that holds a pooled httpx.Client.

with AlgentaClient(api_key=os.environ["ALGENTA_API_KEY"]) as client:
    print(client.get_contract().contract_version)

The constructor accepts four arguments.

Argument	Type	Default	Notes
api_key	str | None	None	Falls back to ALGENTA_API_KEY, then DE_API_KEY.
base_url	str | None	https://api.algenta.ai	Point at your self-hosted engine, e.g. http://localhost:8000.
timeout	float	120.0	Request timeout in seconds.
max_retries	int	3	Retries on 429 and 5xx.

Advanced: contract constants

The package re-exports the platform contract so you can read endpoint paths and defaults without hard-coding them.

from decision_engine import (
    DEFAULT_BASE_URL,             # "https://api.algenta.ai"
    CONTRACT_VERSION,             # the platform contract version this SDK speaks
    PRIMARY_DATA_QUERY_CONTRACT,  # dict: endpoints + query shape
    AUTH_SCHEME,                  # "bearer_api_key"
    API_KEY_PREFIX_LIVE,          # "de_live_"
    API_KEY_PREFIX_TEST,          # "de_test_"
)

print(PRIMARY_DATA_QUERY_CONTRACT["api"]["contract_endpoint"])  # /v1/meta/contract

Datasets and the data contract

list_datasets

datasets = client.list_datasets(search="orders", compact=True)
for ds in datasets.datasets:
    print(ds.dataset_id, ds.name)

Parameter	Type	Default	Notes
page	int	1	1-based page index.
limit	int	200	Page size.
search	str | None	None	Free-text match over dataset names.
status	str | None	None	Filter by ingestion status.
source_name	str | None	None	Filter by source.
compact	bool	False	Drop heavy fields for a lighter list.

For large catalogs, iter_datasets(...) yields across pages and accepts a max_items cap.

get_dataset_summary

summary = client.get_dataset_summary(datasets.datasets[0].dataset_id)
print(summary.dataset_id, summary.row_count)

get_contract

get_contract() returns the live platform contract. If the deployment predates the /v1/meta/contract endpoint, the client falls back to deriving the contract from the OpenAPI document.

contract = client.get_contract()
print(contract.contract_version)

Querying data

The query surface targets the /v1/query family. Requests are plain dicts; you can also pass keyword arguments, which are merged into the request body.

query_with_metadata

Returns the query result plus execution metadata (request id, timing) and the raw response headers.

Python

result = client.query_with_metadata(
    {
        "dataset_id": summary.dataset_id,
        "metric": {"hint": "completed_order_count"},
        "aggregation": "sum",
    }
)
print(result.metadata.request_id)
print(result.data)

cURL

curl -sS https://api.algenta.ai/v1/query \
  -H "Authorization: Bearer $ALGENTA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "ds_orders",
    "metric": {"hint": "completed_order_count"},
    "aggregation": "sum"
  }'

query_batch

Send several queries in one round trip to /v1/query/batch.

batch = client.query_batch(
    {
        "queries": [
            {"dataset_id": summary.dataset_id, "metric": {"hint": "revenue"}, "aggregation": "sum"},
            {"dataset_id": summary.dataset_id, "metric": {"hint": "orders"}, "aggregation": "count"},
        ]
    }
)

query_sql_report

Run a structured SQL-style report against /v1/query/sql-report.

report = client.query_sql_report(
    {
        "dataset_id": summary.dataset_id,
        "select": ["product_line", "sum(revenue) AS revenue"],
        "group_by": ["product_line"],
        "order_by": [{"field": "revenue", "direction": "desc"}],
        "limit": 10,
    }
)

Typed query filters

QueryFilterSpec and QueryFilterCondition are exported for building filters with types rather than raw dicts.

Python

from decision_engine import QueryFilterSpec, QueryFilterCondition

filters = QueryFilterSpec(
    conditions=[
        QueryFilterCondition(field="status", operator="eq", value="completed"),
        QueryFilterCondition(field="region", operator="in", value=["EU", "UK"]),
    ]
)

result = client.query_with_metadata(
    {
        "dataset_id": summary.dataset_id,
        "metric": {"hint": "revenue"},
        "aggregation": "sum",
        "filters": filters.model_dump(exclude_none=True),
    }
)

cURL

curl -sS https://api.algenta.ai/v1/query \
  -H "Authorization: Bearer $ALGENTA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "ds_orders",
    "metric": {"hint": "revenue"},
    "aggregation": "sum",
    "filters": {"conditions": [{"field": "status", "operator": "eq", "value": "completed"}]}
  }'

Decisions and simulation

These methods target the decision and simulation endpoints. simulate and recommend take keyword arguments; plan_decision takes a request dict.

# /v1/simulate — returns a DecisionEnvelope
envelope = client.simulate(
    mode="auto",
    scenario={
        "variables": {"revenue": {"low": 80000, "high": 200000}},
        "objective": "maximize_net_value",
    },
)
print(envelope.recommended_action, envelope.confidence)

# /v1/decisions/plan — returns a DecisionPlanResult
plan = client.plan_decision(
    {
        "goal": "Reduce checkout latency",
        "context": {"p95_ms": 820, "budget_usd": 5000},
    }
)

# /v1/recommend — ranks candidate actions
ranking = client.recommend(
    actions=[{"name": "scale_up"}, {"name": "add_cache"}],
    objective="maximize_net_value",
)

verify (POST /v1/verify) and explain (which runs a query and annotates it) complete the decision surface:

verdict = client.verify({"dataset_id": summary.dataset_id, "claim": {"metric": "revenue", "op": "gt", "value": 0}})
trace = client.explain({"dataset_id": summary.dataset_id, "metric": {"hint": "revenue"}, "aggregation": "sum"})

Capability plane

The capability plane lets an agent discover, route, and execute capabilities (tools, skills, MCP-backed providers).

# List providers, then narrow to MCP-backed ones.
providers = client.list_capability_providers()
mcp = client.list_mcp_providers()      # providers whose provider_type == "mcp_provider"

# Catalog of capabilities; list_skills() is list_capabilities(kinds=["skill"]).
skills = client.list_skills()
cap = client.get_capability("cap.skill.summarize", include_instruction=True)

# Plan a route for a task, then execute the chosen capability.
route = client.route_capabilities({"task": "summarize the incident", "kinds": ["skill"]})
outcome = client.execute_capability(
    {
        "capability_id": cap.capability_id,
        "input": {"text": "..."},
    }
)

Capability plane method reference

Method	Endpoint	Returns
list_capability_providers()	GET /v1/capability-providers	list of CapabilityProviderResult
get_capability(capability_id, include_instruction=False)	GET /v1/capabilities/{id}	CapabilityCatalogEntryResult
list_capabilities(kinds=, provider_ids=, binding_ids=)	GET /v1/capabilities	catalog list
route_capabilities(request)	POST /v1/capabilities/route	CapabilityRoutePlanResult
execute_capability(request)	POST /v1/capabilities/execute	CapabilityExecutionResult
list_skills()	GET /v1/capabilities?kinds=skill	skill catalog
list_mcp_providers()	filtered providers	MCP providers

Agent runs

Create a run, then poll its state and event log.

run = client.create_agent_run(
    task="Triage the failing checkout test and propose a fix",
    context={"repository_id": "repo_42"},
    tools=["repository.triage", "repository.simulate"],
    max_steps=10,
    approval_mode="auto",     # or "manual"
    start_paused=False,
)

state = client.get_agent_run(run.run_id)
events = client.get_agent_run_events(run.run_id, limit=1000)
for event in events.events:
    print(event.type, event.created_at)

create_agent_run posts to /v1/agent/runs. get_agent_run reads /v1/agent/runs/{run_id}; get_agent_run_events reads /v1/agent/runs/{run_id}/events. For a live feed, stream_agent_run_events(run_id) returns an iterator over server-sent events.

Lifecycle controls — approve_agent_run, resume_agent_run, cancel_agent_run, fork_agent_run, and replay_agent_run — round out the surface for human-in-the-loop and reproducibility workflows.

Repository intelligence

The repository methods take a repository_id and a request dict. triage_repository and apply_repository map to the same endpoints the agent uses internally.

# Rank likely-broken files / propose where to act. POST /v1/repositories/{id}/triage
triage = client.triage_repository(
    "repo_42",
    {"goal": "checkout test is red", "max_candidates": 5},
)

# Simulate a proposed change. POST /v1/repositories/{id}/simulate -> DecisionEnvelope
sim = client.simulate_repository(
    "repo_42",
    {"decision_plan_id": triage.decision_plan_id},
)

# Apply the change. POST /v1/repositories/{id}/apply -> RepositoryApplyResult
applied = client.apply_repository(
    "repo_42",
    {"decision_plan_id": triage.decision_plan_id, "confirm": True},
)
print(applied.diff_uri)

apply_repository is a write. The platform contract is read-only by default and write operations require explicit confirmation, so pass the confirmation field your deployment expects rather than relying on defaults.

Async client

Every method above has an async equivalent on AsyncAlgentaClient. The async client is an async context manager and exposes the same surface, split across the async_client_* modules internally.

import asyncio
import os
from decision_engine import AsyncAlgentaClient

async def main() -> None:
    async with AsyncAlgentaClient(api_key=os.environ["ALGENTA_API_KEY"]) as client:
        datasets = await client.list_datasets(search="orders", compact=True)
        summary = await client.get_dataset_summary(datasets.datasets[0].dataset_id)
        result = await client.query_with_metadata(
            {"dataset_id": summary.dataset_id, "metric": {"hint": "revenue"}, "aggregation": "sum"}
        )
        print(result.metadata.request_id)

asyncio.run(main())

Errors

The client raises typed exceptions, all subclasses of DecisionEngineError.

from decision_engine import (
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ServerError,
    ValidationError,
    DecisionEngineError,
)

try:
    client.get_dataset_summary("does-not-exist")
except NotFoundError:
    ...
except RateLimitError:
    ...   # the client already retried 429s up to max_retries
except DecisionEngineError as exc:
    print(exc)

Related pages

Authentication & API keysCLI referenceSelf-hostingAPI endpoint reference