Python SDK

Call tools and models through the MAQPNA gateway from Python, and handle denials, approvals and results.

The MAQPNA Python SDK (maqpna) is the library a Python agent uses to call tools through the MAQPNA gateway. The gateway checks the session token, evaluates policy, runs data loss prevention (DLP) checks, asks a person when a rule requires approval, meters the call and writes it to the audit ledger. The SDK turns each of those outcomes into a typed Python exception.

sequenceDiagram
    participant A as Your agent
    participant S as maqpna SDK
    participant G as MAQPNA gateway
    participant M as MCP server
    A->>S: echo.call("delete_resource", id="db-1")
    S->>G: tools/call (Bearer session token)
    G->>G: identity, policy, DLP, budget, taint
    alt allowed
        G->>M: forward the call
        M-->>G: result
        G-->>S: result
        S-->>A: ToolResult
    else denied
        G-->>S: JSON-RPC -32001 {policy, rule, reason}
        S-->>A: raise PolicyDenied
    else held for approval (async mode)
        G-->>S: JSON-RPC -32002 {approvalId, expiresAt}
        S-->>A: raise ApprovalPending
    end

• Python 3.10 or later. The only runtime dependency is httpx.
• Model Context Protocol (MCP) Streamable HTTP, with JSON and text/event-stream responses.
• The session token is re-read from its file on every request, because the operator rotates it in place. After a 401 the SDK re-reads it and retries once.

Install#

python3 -m venv .venv && . .venv/bin/activate
pip install ./sdk/python                      # core
pip install "./sdk/python[langgraph]"         # with a framework adapter

The extras are langchain, langgraph, openai-agents, claude-agent-sdk and crewai. Each adapter imports its framework only when you use it. See Framework adapters.

Configuration#

MaqpnaClient.from_env() reads the environment the operator gives every sandbox. maqpna dev run sets the same variables on a laptop (see Run your own agent).

Keyword arguments to from_env() override the environment. You can also build the client directly:

from maqpna import MaqpnaClient

client = MaqpnaClient(
    "http://127.0.0.1:8080",
    token="…",                 # or token_file=..., or token_provider=callable
    session="dev-1",
    async_approvals=False,
    timeout=120.0,             # seconds, per HTTP request
)

List and call tools#

from maqpna import MaqpnaClient, PolicyDenied, ApprovalPending, Result

client = MaqpnaClient.from_env()
echo = client.tools("echo")

print("tool servers:", client.available_tool_servers())
for tool in echo.list_tools():          # only the tools policy lets this session see
    print("tool:", tool.name)

print("echo ->", echo.call("echo", text="hello from Python").text)

try:
    echo.call("delete_resource", id="db-1", namespace="kube-system")
except PolicyDenied as e:
    print("denied:", e.rule, e.reason, e.data.get("detail"))

try:
    echo.call_tool("delete_resource", {"id": "db-1", "namespace": "dev"}, async_mode=True)
except ApprovalPending as p:
    print("pending approval:", p.approval_id, "expires", p.data.get("expiresAt"))

print(client.report_result(Result.SUCCEEDED, "echo ok; kube-system delete denied"))

Run it under a local MAQPNA:

maqpna dev up
maqpna dev run -- python agent.py

maqpna dev: session dev-99fe04ae (agent coder, namespace dev, on behalf of you@localhost)
tool servers: ['echo']
tool: echo
tool: get_time
tool: delete_resource
echo -> hello from Python
denied: never-touch-system-namespaces policy_denied system namespaces are off-limits to agents
pending approval: apr_1b11cd074d10313c4cee8c8c expires 2026-10-02T23:54:25.265441-04:00
{'annotated': False, 'at': '2026-10-03T03:49:25.269338Z', 'auditSeq': 4, 'namespace': 'dev', 'redacted': False, 'result': 'Succeeded', 'session': 'dev-99fe04ae', 'summary': 'echo ok; kube-system delete denied'}
maqpna dev: dev-99fe04ae: 1 allow, 1 deny, cost $0.0000 (maqpna dev timeline --last)

The API in brief:

Model calls#

The model route is OpenAI-compatible, so use the official openai package. The session token is the API key; the gateway checks the models:<route> scope, forces the route's model, meters tokens against the budget and writes an audit record (it does not store the prompt).

from openai import OpenAI
from maqpna import MaqpnaClient

client = MaqpnaClient.from_env()
llm = OpenAI(base_url=client.model_endpoint("stub"), api_key=client.token())
reply = llm.chat.completions.create(
    model=client.model_name() or "stub",
    messages=[{"role": "user", "content": "Say hello"}],
)

With maqpna dev up --stub-llm, the scripted model answers with a tool call:

maqpna dev: session dev-3691964c (agent coder, namespace dev, on behalf of you@localhost)
finish: tool_calls
tool calls: [('echo', '{"text":"hello from stub-llm"}')]
content: None
usage: 53
maqpna dev: dev-3691964c: 1 allow, cost $0.0000 (maqpna dev timeline --last)

• client.model_endpoint() with no argument returns MAQPNA_MODEL_ENDPOINT, which the operator sets in a sandbox. maqpna dev run does not set it, so locally pass the route name (model_endpoint("stub")) or use OPENAI_BASE_URL, which maqpna dev run exports when the stack runs with --stub-llm. Without a route, model_endpoint() raises ConfigError; it never falls back to a public API.
• The token rotates in place. For a long-lived OpenAI client, pass extra_headers=client.openai_headers() on each call, or build the client per task.
• Model-route errors are OpenAI-style bodies with error.type (maqpna_unauthorized, maqpna_policy_denied, maqpna_budget_exceeded with HTTP 402, maqpna_approval_required, maqpna_egress_denied, …) and error.reason. See Error and reason codes.

Denials and approvals#

A rule with require_approval holds the call for a human. The SDK supports two modes.

Sync (default). The gateway keeps the request open until an approver decides. The call returns the result, or raises PolicyDenied with reason="approval_denied" (or approval_expired).

Async. The request carries X-Maqpna-Async: 1 (per call with async_mode=True, or for every call with MAQPNA_ASYNC_APPROVALS=1). The gateway answers at once with JSON-RPC -32002, which the SDK raises as ApprovalPending(approval_id). wait_for_approval sends the call in async mode and polls until someone decides:

from maqpna import MaqpnaClient, PolicyDenied, wait_for_approval

client = MaqpnaClient.from_env()
echo = client.tools("echo")
try:
    result = wait_for_approval(
        echo, "delete_resource", {"id": "db-1", "namespace": "dev"},
        poll_interval=2, timeout=120,
        on_pending=lambda p: print(f"waiting for approval {p.approval_id}", flush=True),
    )
    print("approved:", result.text)
except PolicyDenied as e:
    print("denied:", e.reason, e.data.get("detail"))

Approve it from a second terminal with maqpna approvals, the console or MAQPNA Desk (see Human approvals):

eval "$(maqpna dev env)"
maqpna approvals approve apr_13e2aef035de9b2f1258a1eb --approver reviewer@localhost --note "test resource"

maqpna dev: session dev-7c632aec (agent coder, namespace dev, on behalf of you@localhost)
waiting for approval apr_13e2aef035de9b2f1258a1eb
approved: deleted resource db-1 in namespace dev (simulated)
maqpna dev: dev-7c632aec: 1 allow, cost $0.0000 (maqpna dev timeline --last)

Each poll re-sends the same tools/call with X-Maqpna-Approval-Id. An approval lets exactly that one call run; pass approval_id= to resume one queued earlier. Beyond timeout, wait_for_approval raises ApprovalTimeout.

Exceptions#

All inherit from MaqpnaError.

Match on e.reason up to the first : (dlp:pan → dlp). The full list is in Error and reason codes.

Streaming#

• Tool calls: the SDK reads text/event-stream responses from the gateway incrementally and returns the JSON-RPC response that matches the request. It skips server notifications on the stream; it does not expose progress notifications or partial tool output to your code.
• Model calls: stream=True on the OpenAI client works through the model route. The gateway relays the server-sent events and applies DLP to the stream. Real output with the stub model:

{'role': 'assistant', 'tool_calls': [{'index': 0, 'id': 'call_chatcmpl-stub-16_0', 'function': {'arguments': '{"text":"hello from stub-llm"}', 'name': 'echo'}, 'type': 'function'}]} None
{} tool_calls

Report the result#

Tell MAQPNA how the run ended. The gateway runs DLP on the summary, sets status.outcome on the session and appends a session_result audit record.

from maqpna import Result, truncate_utf8

client.report_result(Result.SUCCEEDED, truncate_utf8(final_answer))  # at most 1024 UTF-8 bytes

It raises InvalidRequest for a bad value or an oversized summary, PolicyDenied when DLP blocks the summary (422, reason dlp:<detector>) or the session is revoked, and retries a 503 three times. Under maqpna dev there is no AgentSession resource, so the response shows annotated: False.

Trace context#

Pass a W3C traceparent to link the gateway's spans and the audit record (ext.traceId) to your trace:

client = MaqpnaClient.from_env(traceparent=lambda: current_traceparent())   # or env TRACEPARENT
client.tools("github").call_tool("get_issue", {"number": 1}, traceparent="00-…-01")

Examples#

Next: Framework adapters, Use a framework template, Error and reason codes.