kblock (Python)

A thin Python client for the K-Block Gateway, with LangChain, AutoGen and CrewAI adapters.

The Gateway is an MCP server (Streamable HTTP) in front of your organization's MCP servers. It applies the organization's policy, holds calls for human approval, executes approved calls exactly once and records a signed receipt. This package connects to it with the official MCP Python SDK (Authorization: Bearer <gateway API key>) and turns its answers into typed results and errors. It makes no decisions locally: no policy, approval or security logic runs in the client.

Pre-release (0.1.0rc1) for invited pilot users. You need a K-Block Gateway URL and an agent's gateway API key (kb_live_… / kb_test_…, shown once when it is created in the K-Block dashboard). K-Block gives you both.

Install

Into your agent's own virtual environment:

python3 -m venv .venv                # Windows: python -m venv .venv
source .venv/bin/activate            # Windows (Git Bash): source .venv/Scripts/activate
pip install "kblock==0.1.0rc1"

Framework adapters are extras:

Install	Adds	Python
pip install "kblock[langchain]==0.1.0rc1"	langchain-core 1.x	3.10 or later
pip install "kblock[autogen]==0.1.0rc1"	autogen-core and autogen-agentchat 0.7 (model clients such as autogen-ext are yours to choose)	3.10 or later
pip install "kblock[crewai]==0.1.0rc1"	crewai 1.x, 1.15.9 or later (about 140 packages in the locked resolution)	3.10 to 3.13

Install [autogen] and [crewai] into separate environments: installed with pip after [autogen], [crewai] replaced protobuf with a version that autogen-core 0.7.5 does not accept. Pin the exact version as shown: an unpinned install that allows pre-releases (--pre) would also admit pre-releases of every dependency. Importing an adapter without its extra raises ModuleNotFoundError for the framework.

Requirements: Python 3.10 or later; the MCP Python SDK 1.x (mcp>=1.28,<2; CrewAI 1.15.23, the tested version, pins mcp~=1.28.1), httpx, anyio. The client is async (async with connect_kblock(...)); the CrewAI adapter is synchronous.

A first program

The code on this page is taken unchanged from K-Block's example programs (quickstart.py, agent.py, langchain_agent.py, autogen_agent.py, crewai_agent.py), which K-Block's test suites run against a real Gateway. quickstart.py makes three calls: one the policy allows, one that needs a person's approval, one the policy denies.

import asyncio
import io
import os
import sys
import uuid

from kblock import KBlockApprovalPending, KBlockCompleted, connect_kblock, explain_result

# The Gateway's explanations contain characters such as "≤". On Windows, Python writes output that goes
# to a file or a pipe in the ANSI code page (cp1254, cp1252, ...), which may lack them (UnicodeEncodeError).
# This program writes UTF-8 instead, as PYTHONUTF8=1 would.
for stream in (sys.stdout, sys.stderr):
    if isinstance(stream, io.TextIOWrapper):
        stream.reconfigure(encoding="utf-8")

gateway_url = os.environ.get("KBLOCK_GATEWAY_URL")
api_key = os.environ.get("KBLOCK_API_KEY")
if not gateway_url or not api_key:
    print("Set KBLOCK_GATEWAY_URL and KBLOCK_API_KEY to the Gateway URL and the agent's gateway API key that K-Block gave you.", file=sys.stderr)
    sys.exit(2)


async def main() -> None:
    async with connect_kblock(gateway_url, api_key) as kb:
        # 1. Allowed: the Gateway executes the call exactly once and records a receipt.
        r = await kb.call_tool("lookup_order", {"order_id": f"ord_{uuid.uuid4()}"})
        if isinstance(r, KBlockCompleted):
            print("1. completed:", r.structured_content, "receipt", r.receipt_id, f"({r.receipt_object_id})")
        else:
            print("1.", explain_result(r))

        # 2. Approval required: nothing runs until a person approves the call in the K-Block dashboard;
        #    wait_for_approval repeats the same call until the decision lands, and it then executes once.
        pending = await kb.call_tool("refund", {"order_id": f"ord_{uuid.uuid4()}", "amount": "250.00", "currency": "USD"})
        print("2.", explain_result(pending))
        if isinstance(pending, KBlockApprovalPending):
            print(f"   waiting for a person to approve request {pending.approval_request_id} in the K-Block dashboard")
            done = await kb.wait_for_approval(pending, timeout_seconds=600, wait_seconds_per_request=30)
            if isinstance(done, KBlockCompleted):
                print("   completed:", done.structured_content, "receipt", done.receipt_id, f"({done.receipt_object_id})")
            else:
                print("  ", explain_result(done))

        # 3. Denied: the policy's explanation; nothing was executed.
        denied = await kb.call_tool("refund", {"order_id": f"ord_{uuid.uuid4()}", "amount": "2500.00", "currency": "USD"})
        print("3.", explain_result(denied))


asyncio.run(main())

KBLOCK_GATEWAY_URL="<Gateway URL>" KBLOCK_API_KEY="<gateway API key>" python quickstart.py

The client reads no environment variable itself: your application passes the Gateway URL and the key to connect_kblock(gateway_url, api_key), from wherever it keeps its configuration. The connection lives as long as the async with block. The program assumes an organization whose tools lookup_order (action orders.read) and refund (action refund.issue, with its amount and currency) are under a policy made from the Support refunds template (support-refunds@1, default limits); with other tools, use their names, arguments and amounts. Amounts are decimal strings ("250.00"). The Gateway URL must be https:, or http: to a loopback host; a bad URL raises ValueError. allow_insecure_http=True allows plain http elsewhere (the key would travel in clear text).

Outcomes

await kb.call_tool(name, arguments, *, idempotency_key=None, wait_seconds=None, timeout=None) returns a typed result:

Result (status)	Meaning
KBlockCompleted (completed)	executed; action_id, receipt_id, receipt_object_id, upstream content / structured_content; observe when executed under an observe window
KBlockApprovalPending (approval_pending)	nothing ran; approval_request_id, expires_at; repeat the same call after approval
KBlockDenied (denied)	nothing ran; policy (KBlockPolicyExplanation: policy, rule, and, only when the organization opts in to the conditions disclosure, conditions with thresholds and amount band) and why
KBlockClosed (rejected / expired / cancelled)	a recorded action that will not run
KBlockUnavailable (unavailable)	nothing ran; retryable says whether to retry
KBlockUpstreamError (upstream_error)	the upstream failed; check its effect before repeating
KBlockRefused (invalid_request, plan_limit, replay, unauthorized)	the other refusals
KBlockUnrecognized	a status this version does not know (never a success)

In the policy's terms: a call the policy allows is executed (KBlockCompleted); a call that requires approval is KBlockApprovalPending until a person decides in the K-Block dashboard (Approvals); a call the policy denies is KBlockDenied with the code POLICY_DENIED. denied also covers checks K-Block makes before the policy, each with its own code (for example INVALID_AMOUNT for an amount it cannot read, or NO_APPROVERS when approval is required but no member who may approve has a personal K-Block key). A completed result does not say whether a person approved the call first; its receipt does. observe is set when the call was executed under an observe window, which relaxes the policy's default deny (and, if the owner chose so, an approval rule without a guard) while a new policy is observed. At the organization's disclosure none, policy and why are None.

explain_result turns any outcome into text a model can act on (code, detail, Why:, the policy name@version, rule and conditions). The client fails closed: a success without _meta.kblock evidence (the protected action), or an error without structuredContent.kblock, raises KBlockProtocolError and is never reported as completed. KBlockCompleted.receipt_id is None only when the Gateway executed the call but could not read its receipt back while answering: the call ran (do not repeat it).

Approvals and idempotency

From agent.py: a call that needs a person's approval ran nothing, and wait_for_approval resumes the SAME call (same arguments, same idempotency key) until the decision lands; it then executes once.

    async with connect_kblock(gateway_url, api_key) as kb:
        pending = await kb.call_tool("refund", args)  # with a fresh idempotency key (the default)
        if not isinstance(pending, KBlockApprovalPending):
            fail(explain_result(pending))

        final = await kb.wait_for_approval(pending, timeout_seconds=600, wait_seconds_per_request=30)
        if not isinstance(final, KBlockCompleted):
            fail(explain_result(final))

By default call_tool sends a fresh kblock/idempotency-key, so resume() and wait_for_approval() can never execute a call twice; a plain new call_tool sends a fresh key, which is a new request that needs its own approval. You can pass your own key, or False to send none. Without a key, repeating the identical call resumes the open action by its fingerprint.

Errors

Transport failures raise typed errors, and no message ever contains the key's secret part:

Error	When
KBlockAuthError	HTTP 401 (UNAUTHORIZED), or a malformed key refused before sending (MALFORMED_API_KEY)
KBlockRateLimitedError	HTTP 429, with retry_after_seconds
KBlockUnavailableError	HTTP 503 DATABASE_UNAVAILABLE / GATEWAY_SHUTTING_DOWN (the Gateway's own answers: nothing processed); RPC_UNAVAILABLE (a JSON-RPC answer that the Gateway's database is unavailable, e.g. for list_tools); HTTP_ERROR for other 5xx, including a 503 without one of those codes (a proxy in front of the Gateway: outcome unknown); network errors; timeouts (outcome_unknown says whether the request may have been processed)
KBlockHttpError	413, 404 and other non-MCP answers
KBlockProtocolError	not a K-Block answer (see above)

These errors are raised for the one call that met them, and the connection stays usable. An error a tool call raised carries call (tool, arguments and the idempotency key it was sent with). After an error with outcome_unknown, a call that has a key (e.call.idempotency_key) is repeated safely with await kb.resume(e.call), never as a new call; a call sent without a key would run again, so check its effect first.

tools/call is sent with ClientSession.send_request, not with ClientSession.call_tool. After a successful call, call_tool may issue its own tools/list and validate the output schema; if that failed, the SDK would raise an error for a call the Gateway had already executed and recorded.

Receipts

Each executed call has a receipt (a .kbundle); KBlockCompleted.receipt_id is its ID. A member of the organization downloads it in the K-Block dashboard (the receipt's page, Download .kbundle), or a program downloads it from the K-Block Cloud API with a management API key that has the receipts:read scope (Authorization: Bearer <management API key>); the agent's gateway API key cannot. ReceiptLocation(organization_id, api_url, dashboard_url) in connect_kblock(..., receipts=...) lets kb.receipt_urls(result) build the receipt's URLs. Rejected and cancelled calls also end with a receipt; a call denied when it was decided, and an approval request that expired, end without one.

This package does not include a verifier or a command-line tool. A receipt verifies offline with the npm package @kblockai/sdk (Node.js 22 or later) and the verify_receipt.mjs program from its README, in a directory of its own:

mkdir -p ~/src/kblock-verify && cd ~/src/kblock-verify
npm init -y && npm pkg set type=module
npm install --save-exact @kblockai/sdk@0.1.0-rc.1
# Save the verify_receipt.mjs program from the @kblockai/sdk README in this directory, then:
node verify_receipt.mjs "<path of the .kbundle file>" "<your Gateway's KID>"

verify_receipt.mjs is the complete program in the @kblockai/sdk README: it calls verifyBundleJson and renderText from @kblockai/sdk/verify, prints the seven checks and RESULT: VALID or RESULT: INVALID, and compares the receipt's issuer with your Gateway's KID (K-Block gives it to you), because a valid receipt does not prove on its own which Gateway issued it.

Framework adapters

Each adapter exposes the tools the Gateway lists for the agent. The model reads every outcome as the tool's output (explain_result), and every refusal (held for approval, denied, rejected, expired, unavailable, an upstream that failed…) reaches the framework as a failed tool run, never as a success: a LangChain ToolMessage with status="error", an AutoGen FunctionExecutionResult with is_error=True, a CrewAI ToolFailure. refusals_as_errors=False reports refusals through the framework's success signal instead; throw_on_refusal raises them. Transport errors raise, with one exception: the CrewAI adapter returns a call whose outcome is unknown as a ToolFailure (retryable=False), because CrewAI re-invokes and retries a tool that raised. The CrewAI adapter needs CrewAI's crewai.tools.tool_failure module, which CrewAI added in 1.15.9, so the [crewai] extra requires crewai>=1.15.9,<2 (tested with 1.15.23). With an older CrewAI installed some other way, refusals and unknown outcomes come back as plain text through CrewAI's success path. By default the adapters send no idempotency key, so when the model repeats the identical call after approval, the Gateway resumes the approved action; once that action has finished, an identical call without a key is a new request, decided again. Pass idempotency_key= (a function of the tool name and arguments) to derive a key, so that a repeat never runs twice. Only the fields the model gave are sent (no schema defaults are filled in).

The snippets below are taken from the example programs, where kb is the connected client and model and seen are the program's own.

LangChain (kblock[langchain]; async StructuredTool; ToolMessage.artifact is the typed result), from langchain_agent.py:

from kblock.langchain import KBlockToolRefusal, load_kblock_tools

    async with connect_kblock(gateway_url, api_key) as kb:     # Authorization: Bearer <gateway API key>
        tools = {t.name: t for t in await load_kblock_tools(kb)}

AutoGen (kblock[autogen]; autogen_core BaseTool; an AssistantAgent hands the model explain_result(...)), from autogen_agent.py:

from kblock.autogen import load_kblock_tools

tools = await load_kblock_tools(kb)                     # or load_kblock_tools(kb, tools=["refund"])

    agent = AssistantAgent("support_agent", model_client=model, tools=list(tools), reflect_on_tool_use=False)

CrewAI (kblock[crewai]; sync BaseTool; one short connection per call; the typed result via on_result / last_result), from crewai_agent.py:

from kblock.crewai import KBlockGatewayConfig, load_kblock_tools

seen: list = []
config = KBlockGatewayConfig(gateway_url, api_key)   # sent as Authorization: Bearer <key>; never in repr or model_dump
tools = {t.name: t for t in load_kblock_tools(config, on_result=seen.append)}

Versioning

0.1.0rc1 is a pre-release. Package versions are independent of the K-Block protocol version (v0.1).

License

K-Block SDK License (proprietary); see the LICENSE file in this package.