b24api 2.x

b24api is a thin asynchronous Bitrix24 REST client for Python 3.12+. It knows how to send requests, split logical batches, traverse lists, retry safely, preserve caller correlation and close resources. It does not contain a Tasks, CRM or IM method catalog and does not impose application storage rules.

Install and configure

uv sync --frozen
export BITRIX24_API_WEBHOOK_URL='https://portal.example/rest/.../'

Keep the webhook out of source, logs and command arguments. Reuse one client for a related unit of work so its HTTP/2 connection pool and rate state are reused.

from b24api import Bitrix24, Request

async with Bitrix24() as client:
    profile = await client.call(Request("profile"))

The client owns its default transport. An injected transport remains caller-owned. aclose() is idempotent and closes active streams before the owned transport.

Direct calls

Use call() for detached decoded JSON and call_response() when you also need the immutable response envelope: result, total, next, timing and bounded diagnostic evidence.

Use call_bytes() when a successful method response is a file rather than a Bitrix JSON envelope. The operation is explicit and never hides malformed JSON by falling back to bytes.

archive = await client.call_bytes(Request("example.export.download", replay_safety=ReplaySafety.SAFE))
payload = archive.body

from b24api import ReplaySafety

request = Request("example.item.get", {"id": 7}, ReplaySafety.SAFE)
decoded = await client.call(request)
response = await client.call_response(request)

Replay safety

Request.replay_safety describes what the client may do when a connection fails after the request may already have reached Bitrix:

A failure proved to occur before dispatch may still be retried. Method names never imply safety; mark a request SAFE only when the operation's semantics justify it.

Use ExecutionPolicy to narrow attempts or resource budgets for one operation:

from b24api import ExecutionPolicy

one_attempt = ExecutionPolicy(max_attempts_per_request=1)
result = await client.call(request, policy=one_attempt)

Logical batch and correlation

batch() accepts an arbitrary-length synchronous or asynchronous command source. It consumes the source incrementally and splits it into physical Bitrix batches of at most 50 commands; the full input is never materialized.

Command.correlation is arbitrary caller-owned state. It is retained by reference, returned with the outcome, never serialized to Bitrix and never included in safe diagnostics. This is useful for matching a result to the object, file, chat or database row that produced its request.

from b24api import Command, CommandSuccess

commands = (
    Command(
        Request("example.item.get", {"id": item_id}, ReplaySafety.SAFE),
        correlation=item_id,
    )
    for item_id in source_ids
)

async with client.batch(commands, batch_size=25) as stream:
    async for outcome in stream:
        assert isinstance(outcome, CommandSuccess)
        consume(outcome.correlation, outcome.result)

batch() is fail-fast. batch_outcomes() continues where safe and yields one of CommandSuccess, CommandFailure, CommandNotExecuted or CommandOutcomeUnknown in input order.

from b24api import CommandFailure, CommandNotExecuted, CommandOutcomeUnknown

async with client.batch_outcomes(commands) as stream:
    async for outcome in stream:
        match outcome:
            case CommandSuccess() as success:
                consume(success.correlation, success.result)
            case CommandFailure() | CommandNotExecuted() | CommandOutcomeUnknown():
                handle(outcome)

For independently dispatchable commands, use fan_out() or fan_out_outcomes() with DirectDispatch or BatchDispatch. Delivery order is explicitly READY or INPUT.

Choosing a list operation

The unsuffixed operation is the basic strategy with the fewest endpoint assumptions. Faster or more specialized mechanics have explicit names and explicit preconditions.

page_size is a local decoded-page cap. It is sent to Bitrix only when you provide the endpoint's exact limit_path; the client never guesses method-specific parameter names.

List traversal comparison

Sequential offset

This is the canonical default. It follows the next returned by the server and confirms the end with an empty page. For ordinary counted Bitrix list endpoints, next is an offset for the next LIMIT/OFFSET page, not a keyset cursor. Producing total involves a separate server-side count query in addition to retrieving the page. These database operations are performed inside the same REST request: the client does not issue an additional HTTP call just for the count. This distinction matters for performance: not making a separate HTTP count call does not mean avoiding server-side COUNT work. iter_list does not suppress that work; a returned total is observational and does not control this strategy's completion. Exact database implementation is endpoint-specific.

from b24api import IdentityCoercion, IdentitySpec, ResultSelector

identity = IdentitySpec(
    item_path=("ID",),
    filter_key="ID",
    order_key="ID",
    coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
)

stream = client.iter_list(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
)
async with stream:
    async for item in stream:
        consume(item)

Without identity, successful exhaustion is reported as MECHANICS_ONLY: pagination completed, but the client cannot prove that the portal did not duplicate or substitute rows.

Mapping-backed collections are explicit as well. MAPPING_VALUES yields values from a selected mapping in insertion order; MAPPING_VALUES_OR_EMPTY additionally accepts only an empty terminal sequence and records that degradation in the operation report.

from b24api import ResultCollectionShape

stream = client.iter_list(
    Request("example.dictionary.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    collection_shape=ResultCollectionShape.MAPPING_VALUES,
)
async with stream:
    async for value in stream:
        consume(value)

Counted, physically batched tail

The first direct page must contain an exact filtered total and, when more rows exist, next. The client derives all remaining offsets from the observed head width and sends tail pages through bounded physical batches.

stream = client.iter_list_counted(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
    page_size=50,
    batch_size=50,
)

Use it only when total is exact for the supplied filter and offset pages are stable. Any missing range, overlap, duplicate identity or total contradiction raises IncompleteTraversalError.

Physical batching reduces HTTP exchanges, but does not suppress server-side COUNT in ordinary counted list subrequests. Each command still performs its own offset page retrieval and associated total calculation on the server. Do not confuse batching these commands with a no-count traversal.

No-count keyset

Keyset traversal uses automatic execution by default. Omitting execution is equivalent to AutoKeysetExecution(StableIntegerKeysetContract()): the client captures both ordered boundaries, then selects boundary-only, sequential, range, or partitioned execution from the observed geometry and available policy capacity. Planning completes before any row is emitted, so partial consumption still pays that barrier cost. A sequential selection made by auto is a cost decision; a failed or contradictory fast plan is never silently restarted as sequential.

By using the default, the caller asserts that the endpoint has a stable, unique integer key, honors strict numeric bounds and ordering, and satisfies empty-confirmation completion. Concurrent mutation outside the captured middle is handled by the finishing sweep; mutation inside it is outside this assertion. Pass SequentialKeysetExecution() explicitly when an endpoint cannot satisfy the fast contract or when the previous request-by-request behavior is required.

Static incompatibility with the auto contract raises CapabilityError from the iter_list_keyset(...) call before iteration begins. A portal that accepts but contradicts the declared controls fails before emission with IncompleteTraversalError; auto never restarts that operation silently. KeysetTraversal inside reference traversal remains sequential-only. The terminal report now includes keyset_execution for omitted-execution keyset calls so consumers can see the requested and selected plan.

Use await client.verify_keyset_capability(...) as a development/CI/staging guard on a stable representative fixture. It performs five strict-bound checks and returns only a VERIFIED report; unsupported and inconclusive verdicts raise KeysetCapabilityError. The ordinary iter_list_keyset() remains a separate caller-asserted operation with zero verifier canaries and may emit a partial prefix before a late endpoint contradiction is detected.

Every list operation also accepts an immutable PageAdapter strategy. The adapter synchronously maps selected frozen items using sibling result metadata while preserving cardinality, order and configured identities. The identity adapter is the default and preserves existing JSON output.

An advisory total may only raise an automatic cost estimate and never proves completion. Reports record the selected strategy and reason: unbounded auto continuation has the same ordered_prefix_only assurance as sequential traversal, while bounded plans additionally report caller_asserted_bounds. Runtime traversal does not perform verifier canaries.

from b24api import KeysetSpec, ParameterPath

stream = client.iter_list_keyset(
    Request("example.item.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    identity=identity,
    keyset=KeysetSpec(
        filter_path=ParameterPath(("filter",)),
        order_path=ParameterPath(("order",)),
    ),
)

Dependent cursor

Use a cursor when the next boundary is returned or derived from the previous page, as with many message-list methods.

from b24api import CursorSpec, ParameterPath

stream = client.iter_list_cursor(
    Request("example.message.list", replay_safety=ReplaySafety.SAFE),
    selector=ResultSelector(("items",)),
    cursor=CursorSpec(
        parameter_path=ParameterPath(("LAST_ID",)),
        item_path=("ID",),
        coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
        direction="ascending",
        take="last",
    ),
)

Cursor values must be unique and strictly monotonic. If an endpoint exposes only a non-unique boundary, use an application-owned direct-call workflow or supply a unique tie-breaker.

For multiple parent-bound cursor chains, iter_cursors() keeps cursor progress and correlation isolated per binding while ready pages share the physical batch queue.

See architecture, migration, performance, and endpoint recipes for the complete contracts and selection guidance.

One list method across many parent entities

Binding applies exact parameter updates to a base request and carries parent correlation. The client remains unaware of entity types: a binding can represent a deal, lead, chat or any other caller-defined parent.

from b24api import (
    BatchDispatch,
    Binding,
    ParameterPath,
    ParameterUpdate,
    ReferenceComplete,
    ReferenceItem,
    SequentialTraversal,
)

bindings = (
    Binding(
        summary=f"owner {parent_id}",
        updates=(ParameterUpdate(ParameterPath(("filter", "OWNER_ID")), parent_id),),
        correlation=parent_id,
    )
    for parent_id in parent_ids
)

stream = client.iter_references(
    Request("example.comment.list", replay_safety=ReplaySafety.SAFE),
    bindings,
    traversal=SequentialTraversal(selector=ResultSelector(("items",)), identity=identity),
    dispatch=BatchDispatch(batch_size=25, concurrency=2),
)
async with stream:
    async for event in stream:
        if isinstance(event, ReferenceItem):
            consume(event.correlation, event.item)
        elif isinstance(event, ReferenceComplete):
            record_completion(event.correlation, event.row_count)

For messages across chats, use the same iter_references() shape: each binding updates the chat parameter and carries the chat correlation; choose CursorTraversal when the message endpoint is cursor-based. Identity tracking and completion remain scoped to each binding, so equal child IDs under different parents are not conflated.

from b24api import (
    Binding,
    CursorSpec,
    CursorTraversal,
    DirectDispatch,
    IdentityCoercion,
    ParameterPath,
    ParameterUpdate,
    ResultSelector,
)

chat_bindings = (
    Binding(
        summary=f"chat {chat_id}",
        updates=(ParameterUpdate(ParameterPath(("DIALOG_ID",)), chat_id),),
        correlation={"chat_id": chat_id},
    )
    for chat_id in chat_ids
)

messages = client.iter_references(
    Request("example.message.list", replay_safety=ReplaySafety.SAFE),
    chat_bindings,
    traversal=CursorTraversal(
        selector=ResultSelector(("items",)),
        cursor=CursorSpec(
            parameter_path=ParameterPath(("LAST_ID",)),
            item_path=("ID",),
            coercion=IdentityCoercion.DECIMAL_STRING_INTEGER,
            direction="ascending",
            take="last",
        ),
    ),
    dispatch=DirectDispatch(concurrency=4),
)

iter_reference_outcomes() additionally yields correlated ReferenceFailure, ReferenceNotExecuted and ReferenceOutcomeUnknown. A malformed source object that is not a Binding has no valid caller correlation, so it terminates the source with InputSourceError rather than fabricating a reference outcome. Already accepted bindings retain their real outcomes.

Streams, partial results and reports

Every multi-item operation returns an OperationStream. Prefer async with: a plain break does not close an arbitrary async iterator. After cleanup, stream.report permanently exposes one immutable OperationReport; before termination it is None.

first = await client.iter_list(request).first()
page = await client.iter_list(request).collect(limit=100)

assert first.report.partial
assert page.report.partial

Helpers do not pull an extra row just to prove exhaustion. Reaching a requested limit is therefore EARLY_CLOSED, never a false COMPLETED. Cancellation and cleanup preserve the primary exception and publish the same final report where the Python exception type permits it.

Resource boundaries

ExecutionPolicy bounds requests, pages, elapsed time, attempts, decompressed response bytes, buffered commands and rows, direct concurrency and active references. The default response ceiling is 16 MiB and is enforced while streaming, before JSON decoding.

Sequential and counted exact traversal retain observed identities in memory. There is no database, spill file or identity-count refusal. Crossing 100,000 distinct identities emits one RuntimeWarning; exact tracking continues. Strict keyset and cursor traversal retain only monotonic progression state when sufficient.

CLI

The wheel installs b24api. Stdout contains only result data; list rows are JSONL. Reports and safe errors go to stderr. Credentials come only from Settings and cannot be passed as CLI arguments.

b24api call profile
b24api call example.item.get --params '{"id":7}' --raw --replay-safety safe
b24api list example.item.list --params @params.json
b24api list example.item.list --strategy counted --contract @counted-contract.json

The --raw CLI option selects the response envelope; it does not alter the Python API. Advanced list strategies use closed JSON version: 1 contracts. The entire contract is validated before client construction. Run b24api --help and b24api list --help for the compact option surface.

Exit codes are 0 success, 2 usage/contract error, 3 unavailable configuration, 4 remote/protocol/correctness/incomplete failure, 5 broken output consumer and 130 cancellation.

Correctness boundaries

The client fails closed on contradictory pagination, missing counted ranges, duplicate identities, unsafe ambiguous replay, oversized responses and incomplete cleanup. It can prove only facts visible through the transport contract: continuation, totals, identity, order, budgets and lifecycle.

It cannot generically prove that Bitrix honored the business meaning of a filter, choose an application's composite storage key or reconcile an ambiguous write. Applications must validate expected business sets and verify writes where needed.

Performance and profiling

The current deterministic profile covers request counts, wall/CPU time, time to first row, high-water counters, retained resources and optional Memray allocations:

uv run python tools/b24api_evidence/profile_runtime.py --capability-suite
uv run python tools/b24api_evidence/profile_runtime.py --samples 7 --warmups 2
uv run --with memray python tools/b24api_evidence/profile_runtime.py \
  --case dense-10k --plan counted_batch --samples 7 --warmups 2 \
  --memray-output /tmp/b24api.bin
uv run --with memray memray stats /tmp/b24api.bin

These deterministic fixtures characterize local resources and network shape; they are not live portal latency admission. See docs/performance.md for current measurements and docs/architecture.md for guarantees and ownership boundaries.

Projects moving from an earlier API surface can use docs/migration.md.

Verification

uv sync --frozen
.venv/bin/pytest -q -p no:cacheprovider
.venv/bin/ruff check . --no-fix --no-cache
.venv/bin/ruff format --check . --no-cache
.venv/bin/mypy --strict b24api tools/b24api_evidence
git diff --check

The wheel regression installs into an isolated environment, executes the b24api entry point and checks that tests, live/evidence tooling and credentials are excluded.