Python

The kblock package connects a Python agent to the Gateway. It mirrors the TypeScript client: it connects with the official MCP Python SDK, sends the agent's gateway API key and the Gateway's _meta conventions, and turns every answer into a typed result or a typed error. It evaluates nothing locally; the Gateway decides (policy, approval, exactly-once execution, signed receipts).

Three optional extras add thin adapters for LangChain, AutoGen and CrewAI, so that a refusal reaches the framework as a tool error while the model still reads K-Block's explanation.

Install

pip install "kblock==0.1.0rc1"

With an adapter, install the matching extra instead:

pip install "kblock[langchain]==0.1.0rc1"
pip install "kblock[autogen]==0.1.0rc1"
pip install "kblock[crewai]==0.1.0rc1"

K-Block's own test environment pins mcp 1.28.1, langchain-core 1.6.5, autogen-core / autogen-agentchat 0.7.5 and crewai 1.15.23.

You need the Gateway URL and a gateway API key for the agent, as described in TypeScript client → Install. The examples on this page read them from two environment variables of their own (KBLOCK_GATEWAY_URL, KBLOCK_API_KEY); your application can pass them from anywhere.

Connect and call a tool

connect_kblock is an async context manager. It checks the key's format and the URL, opens the MCP connection and yields a KBlockClient; leaving the block closes it.

import asyncio
import os

from kblock import KBlockApprovalPending, KBlockCompleted, KBlockDenied, connect_kblock, describe_policy, explain_result


async def main() -> None:
    gateway_url = os.environ.get("KBLOCK_GATEWAY_URL", "")  # e.g. https://gateway.example.com
    api_key = os.environ.get("KBLOCK_API_KEY", "")  # the agent's gateway API key
    async with connect_kblock(gateway_url, api_key) as kb:
        tools = await kb.list_tools()  # only the tools the agent's delegation grants
        print([t.name for t in tools])

        # For example, a refund tool that your policy may allow, hold for approval or deny.
        r = await kb.call_tool("refund", {"order_id": "ord_123", "amount": "250.00", "currency": "USD"})
        if isinstance(r, KBlockCompleted):
            print("result:", r.structured_content or r.content, "receipt:", r.receipt_id)
        elif isinstance(r, KBlockApprovalPending):
            print(f"approval request {r.approval_request_id} is open until {r.expires_at}")
        elif isinstance(r, KBlockDenied):
            print("denied:", r.code, r.detail)
            if r.policy is not None:
                print("decided by", describe_policy(r.policy))
        else:
            print(explain_result(r))  # rejected, expired, cancelled, unavailable, upstream_error, ...


asyncio.run(main())

The rules are the same as in TypeScript:

• A key that is not in the K-Block format raises KBlockAuthError (MALFORMED_API_KEY) before anything is sent.
• The URL must be http or https without credentials, query or fragment; /mcp is appended when missing. Plain http is allowed only to loopback hosts unless you pass allow_insecure_http=True. A bad URL raises ValueError.
• The key is sent only to the Gateway endpoint's origin, and redirects are never followed.
• Numbers in the arguments must be integers, so pass decimals such as amounts as strings ("250.00"). A non-integer number is refused as invalid_request with code INVALID_ARGUMENTS.

Results are frozen dataclasses with snake_case fields; test them with isinstance. The classes are KBlockCompleted, KBlockApprovalPending, KBlockDenied, KBlockClosed (rejected, expired, cancelled), KBlockUnavailable, KBlockUpstreamError, KBlockRefused (invalid_request, plan_limit, replay, unauthorized) and KBlockUnrecognized. Each has a status with the same values as in TypeScript → Outcomes. A completed call has content, structured_content, action_id, receipt_id, receipt_object_id, result_object_id, observe, call and raw; a refusal has code, detail, message, action_id, approval_request_id, expires_at, retryable, receipt, policy, why, call and raw. is_kblock_result(value) tells whether a value is one of them.

Parsing fails closed: an answer without K-Block evidence raises KBlockProtocolError, and an unknown refusal status becomes KBlockUnrecognized, never a success.

Idempotency keys

call_tool(name, arguments, *, idempotency_key=None, wait_seconds=None, timeout=None) sends a fresh key (kbc-<uuid>) by default, like the TypeScript client.

A new call_tool after an approval is therefore a new request. Resume a pending call with wait_for_approval or resume. A key works as in TypeScript: the same key with different arguments is refused with IDEMPOTENCY_KEY_REUSED, and a key whose action ended keeps returning that outcome (see Idempotency keys).

Approvals

wait_for_approval repeats the pending call with kblock/wait-seconds, so the Gateway holds each request until the approver decides (in the dashboard: Approvals, open the request, then Approve and sign).

import asyncio
import os

from kblock import KBlockApprovalPending, KBlockCompleted, connect_kblock, explain_result


async def main() -> None:
    async with connect_kblock(os.environ.get("KBLOCK_GATEWAY_URL", ""), os.environ.get("KBLOCK_API_KEY", "")) as kb:
        first = await kb.call_tool("refund", {"order_id": "ord_123", "amount": "250.00", "currency": "USD"})
        if isinstance(first, KBlockApprovalPending):
            print(f"Approve request {first.approval_request_id} in the K-Block dashboard")
            done = await kb.wait_for_approval(
                first,
                timeout_seconds=600,  # give up after 10 minutes (default 300)
                wait_seconds_per_request=30,  # how long the Gateway may hold each request (default 60)
                on_pending=lambda p: print("still pending, open until", p.expires_at),
            )
            if isinstance(done, KBlockCompleted):
                print("executed once, receipt", done.receipt_id)
            else:
                print(explain_result(done))  # rejected, expired, cancelled... or still pending after 600 s


asyncio.run(main())

• It returns the first result that is not approval_pending, or the last pending result when timeout_seconds passed (it does not raise).
• Rate limits and a temporarily unavailable Gateway (KBlockUnavailableError other than HTTP_ERROR) are waited out, using Retry-After (2 seconds when absent), only when the call has an idempotency key. Without a key, or when the pause would run past timeout_seconds, they are raised.
• resume(result, *, wait_seconds=None, timeout=None) repeats the call once, for your own polling loop.
• wait_seconds must be a whole number from 0 to 600 (ValueError otherwise); the Gateway caps it at its own maximum, 60 seconds by default.
• There is no cancellation parameter: cancel the surrounding task to stop waiting.

Receipts

Pass receipts=ReceiptLocation(organization_id=..., api_url=..., dashboard_url=...) to connect_kblock, then kb.receipt_urls(result) returns a ReceiptUrls with bundle, api and page (or None when the result has no receipt). Without a location it raises ValueError. receipt_id_of(result) and action_id_of(result) read the IDs of any result. api_url is the Cloud API's base URL for programs, which K-Block gives you (see TypeScript client → Receipts).

Downloading the bundle needs a signed-in member of the organization (any role) or a management API key of the organization with the receipts:read scope; the agent's gateway key cannot read it.

Errors

The error classes match the TypeScript client, with snake_case attributes. All extend KBlockError (with code and message); is_kblock_error(e) tests for them.

Invalid options (bad URL, empty tool name, wait_seconds, wait_seconds_per_request, idempotency_key, timeout_seconds, request_timeout, receipt_urls without a location) raise ValueError. When the MCP SDK's task groups wrap a single error, connect_kblock raises that error itself, not an exception group.

import asyncio
import os

from kblock import KBlockAuthError, KBlockHttpError, KBlockProtocolError, KBlockRateLimitedError, KBlockUnavailableError
from kblock import connect_kblock, is_kblock_error


async def main() -> None:
    try:
        async with connect_kblock(os.environ.get("KBLOCK_GATEWAY_URL", ""), os.environ.get("KBLOCK_API_KEY", "")) as kb:
            r = await kb.call_tool("lookup_order", {"order_id": "ord_123"}, idempotency_key="lookup-ord_123")
            print(r.status)
    except KBlockAuthError as e:
        print("key refused:", e.code, e.status)
    except KBlockRateLimitedError as e:
        print("rate-limited; retry after", e.retry_after_seconds)
    except KBlockUnavailableError as e:
        # outcome_unknown: the call may have run. Repeat it only with the same idempotency key.
        print("unavailable:", e.code, "outcome unknown:", e.outcome_unknown)
    except KBlockHttpError as e:
        print("HTTP", e.status, e.message)
    except KBlockProtocolError as e:
        print("not a K-Block Gateway answer:", e.message)
    except Exception as e:
        if not is_kblock_error(e):
            raise
        print(e)


asyncio.run(main())

Differences from the TypeScript client

Decision only: authorize_action

For a system that is not an MCP server, authorize_action (async) and authorize_action_sync ask the Gateway to authorize one exact action that your code then performs itself (POST /v1/authorize; no MCP connection):

import asyncio
import os

from kblock import authorize_action


async def main() -> None:
    answer = await authorize_action(
        os.environ["KBLOCK_GATEWAY_URL"],
        os.environ["KBLOCK_API_KEY"],
        "issue_refund",
        {"order_id": "ord_1001", "amount": "40.00", "reason": "damaged"},
        idempotency_key="refund-ord_1001",
        wait_seconds=30,
    )
    if answer.allowed:
        print("refund now, once; authorization", answer.protected_action_id)
    else:
        print("not authorized:", answer.status, answer.code, answer.detail)


asyncio.run(main())

Both take idempotency_key (None for a fresh one, False for none), wait_seconds, timeout (seconds, on top of the wait) and allow_insecure_http. The result is a KBlockAuthorized (allowed is True) or a KBlockNotAuthorized (status, code, detail, retryable); transport failures raise the same errors as connect_kblock. See Decision-only integrations.

LangChain: kblock[langchain]

kblock.langchain turns the Gateway's tools into async LangChain StructuredTools with the Gateway's input schema and response_format="content_and_artifact": the content is explain_result(result), the artifact the typed result.

import asyncio
import os

from kblock import KBlockApprovalPending, connect_kblock
from kblock.langchain import KBlockToolRefusal, load_kblock_tools


async def main() -> None:
    async with connect_kblock(os.environ.get("KBLOCK_GATEWAY_URL", ""), os.environ.get("KBLOCK_API_KEY", "")) as kb:
        tools = await load_kblock_tools(kb)  # pass `tools` to your async LangChain agent
        refund = next(t for t in tools if t.name == "refund")

        # What an agent executor does with a model's tool call:
        args = {"order_id": "ord_123", "amount": "250.00", "currency": "USD"}
        msg = await refund.ainvoke({"type": "tool_call", "id": "call_1", "name": refund.name, "args": args})
        print(msg.status, type(msg.artifact).__name__)  # "error" KBlockApprovalPending when the call is held
        print(msg.content)  # the explanation the model reads
        if isinstance(msg.artifact, KBlockApprovalPending):
            print("approval request", msg.artifact.approval_request_id)

        # For code that runs the tools itself: raise refusals instead.
        strict = (await load_kblock_tools(kb, tools=["refund"], throw_on_refusal=True))[0]
        try:
            await strict.ainvoke(args)
        except KBlockToolRefusal as e:  # also a LangChain ToolException
            print("refused:", e.code, e.result.status)


asyncio.run(main())

• The tools are async only (the MCP session belongs to the event loop that opened it): use ainvoke or an async agent.
• Invoked with a tool call, a refusal is a ToolMessage with status="error"; refusals_as_errors=False reports it with status="success". Invoked with plain arguments, the tool returns the text only.
• throw_on_refusal=True (or a predicate) raises KBlockToolRefusal, a KBlockRefusalError that is also a LangChain ToolException, so handle_tool_error applies.
• No idempotency key is sent by default: the model's identical repeat resumes the open action by its fingerprint. idempotency_key=lambda name, args: ... derives one per call. wait_seconds lets the Gateway hold each call.
• load_kblock_tools(kb, tools=[...]) raises ValueError for a name the Gateway does not list. kblock_tool(kb, definition, ...) wraps a single tool definition.
• A call whose outcome is unknown (a connection that broke after sending, a timeout, a proxy's 5xx, an answer that is not a K-Block result), invoked with a tool call, is a ToolMessage with status="error" whose text tells the model that the action may have run and must not be repeated blindly; its artifact is {"kblock_status": "outcome_unknown", "code": ..., "tool": ..., "idempotency_key": ...}. Invoked with plain arguments, the error is raised.
• Other transport errors always raise.

AutoGen: kblock[autogen]

kblock.autogen turns the Gateway's tools into AutoGen tools (autogen_core.tools.BaseTool) for an AssistantAgent. It does not use autogen_ext.tools.mcp, which would turn K-Block refusals into plain error text.

import os

from autogen_agentchat.agents import AssistantAgent
from autogen_core.models import ChatCompletionClient

from kblock import connect_kblock
from kblock.autogen import load_kblock_tools


async def run(model_client: ChatCompletionClient) -> None:
    """Call with your model client, for example one from autogen-ext."""
    async with connect_kblock(os.environ.get("KBLOCK_GATEWAY_URL", ""), os.environ.get("KBLOCK_API_KEY", "")) as kb:
        tools = await load_kblock_tools(kb)  # or load_kblock_tools(kb, tools=["refund"])
        agent = AssistantAgent("support_agent", model_client=model_client, tools=list(tools))
        result = await agent.run(task="Refund order ord_123: 250.00 USD.")
        print(result.messages[-1])

• What the model reads is explain_result(result) (the tool's return_value_as_string).
• A refusal reaches AutoGen as a tool error: run_json, which AssistantAgent calls, raises KBlockRefusalError with the explanation as its message, so FunctionExecutionResult.is_error is True. refusals_as_errors=False restores the success flag.
• run(args, cancellation_token) returns the typed result; with throw_on_refusal it raises too.
• For a call whose outcome is unknown, run_json raises KBlockOutcomeUnknownError (code OUTCOME_UNKNOWN; error is the original error) whose message tells the model that the action may have run and must not be repeated blindly.
• Only the arguments the model gave are sent (no schema defaults filled in), so the call's fingerprint matches what the model asked for.
• No idempotency key by default; idempotency_key and wait_seconds work as in the LangChain adapter.

CrewAI: kblock[crewai]

CrewAI calls tools synchronously and from its own threads, while an MCP session belongs to the event loop that opened it. So a CrewAI tool holds connection settings (KBlockGatewayConfig) instead of a live client, and every call opens a short connection to the Gateway, calls the tool and closes it. It does not use crewai-tools' MCP adapter, which would turn K-Block refusals into plain error text.

import os

from crewai import Agent

from kblock import KBlockCallResult
from kblock.crewai import KBlockGatewayConfig, load_kblock_tools

config = KBlockGatewayConfig(os.environ.get("KBLOCK_GATEWAY_URL", ""), os.environ.get("KBLOCK_API_KEY", ""))
outcomes: list[KBlockCallResult] = []
tools = load_kblock_tools(config, on_result=outcomes.append)  # synchronous; aload_kblock_tools in async code

support = Agent(
    role="Support agent",
    goal="Resolve refund requests within the refund policy",
    backstory="You handle customer refunds through K-Block-protected tools.",
    tools=tools,
)

• What the agent reads is explain_result(result); the typed result goes to on_result and to the tool's last_result.
• A refusal is returned as a CrewAI ToolFailure (from crewai.tools.tool_failure) with the K-Block code, so CrewAI records it as a failed tool run and applies its tool failure policy, while the agent still reads the explanation. refusals_as_errors=False returns plain text; throw_on_refusal raises KBlockRefusalError.
• Errors that say nothing was processed (authentication, rate limit, the Gateway's own 503, a connection that never sent the call) raise.
• A call whose outcome is unknown (a connection that broke after sending, a timeout, a proxy's 5xx, an answer that is not a K-Block result) is not raised, because CrewAI retries a tool that raised and a keyless repeat could run the action again. It is returned as a ToolFailure with retryable=False and details["kblock_status"] == "outcome_unknown"; last_result is then None.
• KBlockGatewayConfig(gateway_url, api_key, request_timeout=120.0, allow_insecure_http=False) keeps the key out of repr, and the tools keep it out of their fields and model_dump.
• No idempotency key by default; idempotency_key and wait_seconds work as in the other adapters. load_kblock_tools(config, tools=[...]) raises ValueError for an unknown name.

Next steps