MAQPNA Python SDK (maqpna)

The Python SDK for agents that run under MAQPNA, in a sandbox or under maqpna dev. It calls MCP tools through the MAQPNA gateway, which checks the session token, applies policy, asks a person when a rule requires approval, meters each call and writes it to the audit ledger.

• Python ≥ 3.10. Its only dependency is httpx.
• MCP Streamable HTTP (JSON-RPC 2.0), with both application/json and text/event-stream responses.
• The identity token is re-read from the mounted file on every request, because the operator rotates it in place. After a 401 the SDK retries once with the token it reads again.
• Governance outcomes raise typed exceptions: PolicyDenied, ApprovalPending and Unauthorized.
• client.report_result("Succeeded" | "Failed", summary) tells MAQPNA how the run ended.
• Optional adapters for LangChain, LangGraph, the OpenAI Agents SDK, the Claude Agent SDK and CrewAI.

Install

Published on PyPI as kosa8-maqpna; the package you import is maqpna.

pip install kosa8-maqpna                  # core
pip install "kosa8-maqpna[langchain]"     # + LangChain adapter
pip install "kosa8-maqpna[langgraph]"     # or openai-agents, claude-agent-sdk, crewai

Environment (injected by the MAQPNA operator)

Variable	Meaning
MAQPNA_GATEWAY_URL	Gateway base URL, e.g. http://maqpna-gateway.maqpna-system.svc:8080
MAQPNA_SESSION	The AgentSession name
MAQPNA_TOKEN_FILE	Path to the identity JWT (default /var/run/maqpna/token), rotated in place
MAQPNA_TOOL_<NAME>_URL	{gateway}/mcp/<name> per tool. NAME is upper-cased and each non-alphanumeric character becomes _
MAQPNA_TOKEN	(local dev only) static token used when no token file is available
MAQPNA_ASYNC_APPROVALS	1 sends X-Maqpna-Async: 1 on every call (see below)
MAQPNA_MODEL_ENDPOINT	OpenAI-compatible base URL of the governed model route, {gateway}/llm/{route}/v1
MAQPNA_MODEL_NAME	Model served by that route (the gateway forces it)

Usage

from maqpna import MaqpnaClient, PolicyDenied, ApprovalPending, Unauthorized

client = MaqpnaClient.from_env()
github = client.tools("github")          # -> ToolServer for MAQPNA_TOOL_GITHUB_URL

github.initialize()                      # optional; done lazily on first request
for tool in github.list_tools():         # filtered by policy (FR-3.6)
    print(tool.name, tool.input_schema)

try:
    result = github.call("get_issue", owner="acme", repo="api", number=4821)
    print(result.text)                   # also: result.content, result.structured, result.is_error
except PolicyDenied as e:
    print("blocked by rule", e.rule, "because", e.reason)
except Unauthorized:
    print("token lacks the tools:github scope")

call(name, /, **args) takes the tool arguments as keyword arguments. To pass a dict, or to set options for a single call, use call_tool(name, arguments, async_mode=..., approval_id=...).

Human approval

A require_approval rule has two modes:

• Sync (default): the gateway holds the request open until an approver decides. The call then returns the result, or raises PolicyDenied when the approver denies it.
• Async: the request carries X-Maqpna-Async: 1. The gateway answers right away 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 wait_for_approval, ApprovalTimeout

k8s = client.tools("kubernetes")
try:
    result = wait_for_approval(
        k8s, "delete_pod", {"namespace": "staging", "name": "web-1"},
        poll_interval=5, timeout=900,
        on_pending=lambda p: print(f"waiting for approval {p.approval_id}"),
    )
except PolicyDenied as e:      # the approver said no
    ...
except ApprovalTimeout:
    ...

Each poll sends the same tools/call again with X-Maqpna-Async: 1 and X-Maqpna-Approval-Id: <id>. Once the approval is granted, the gateway runs the call and returns the result.

For polling to work, the gateway must answer a retry whose approval is still pending with -32002. See docs/components-sdk.md §2.

Model (LLM) calls

With egress: gateway-only the agent reaches its model through the gateway as well. The model route is OpenAI-compatible, so you use the official openai client and pass the identity token as the API key. The gateway checks the models:{route} scope, forces the route's model, meters tokens against the session budget and audits each call (the prompt itself is never stored).

from openai import OpenAI

llm = OpenAI(base_url=client.model_endpoint(), api_key=client.token())
reply = llm.chat.completions.create(
    model=client.model_name() or "default",
    messages=[{"role": "user", "content": "Summarise issue 4821"}],
)

The token rotates in place. For long-lived clients, pass fresh headers on each call with extra_headers=client.openai_headers(), or build the OpenAI client per task. Errors come back OpenAI-style, with error.type set to maqpna_unauthorized, maqpna_budget_exceeded (HTTP 402), maqpna_policy_denied, maqpna_approval_required or maqpna_egress_denied.

LangChain

from maqpna.integrations.langchain import as_tools

tools = as_tools(client.tools("github"), prefix=True, wait_for_approvals=False)

Each MCP tool becomes a StructuredTool that uses the tool's JSON schema. When a call is denied by policy, is waiting for approval, is not authorized or cannot reach the gateway, the adapter raises a ToolException. The model sees that message and can change its plan, and the agent keeps running. Calls are sent in async approval mode, so a pending approval is reported at once (a retry of the same call resumes it) instead of holding the request open until the HTTP client times out.

Result reporting

from maqpna import Result, truncate_utf8

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

The gateway (POST /v1/sessions/self/result) DLP-scans the summary, annotates the AgentSession (status.outcome) and appends a session_result audit record. A bad result or an oversized summary raises InvalidRequest (a ValueError); a summary blocked by DLP (422) or a revoked session raises PolicyDenied (reason is dlp:<detector> / revoked); a 503 is retried 3 times.

Framework adapters

Module	Gives you
maqpna.integrations.langgraph	tool_node(servers): a ToolNode whose pending approvals become interrupt() with the approval id; resume polls it. invoke_and_report(client, graph, input, config) reports the run.
maqpna.integrations.openai_agents	mcp_server(client, "github") (MCPServerStreamableHttp, token per request) and function_tools(server, approvals="error" | "wait" | "interrupt") with a needs_approval hook.
maqpna.integrations.claude_agent_sdk	mcp_servers(client) config dict, sdk_mcp_server(...) in-process proxy, MaqpnaPermissions hooks / can_use_tool (-32001 → deny, -32002 → ask), query_and_report(...).
maqpna.integrations.crewai	as_tools(server): CrewAI BaseTools; denials and pending approvals come back as text.

Each adapter imports its framework only when used. See docs/components-sdk.md §8.

MCP 2026-07-28 and trace context

MaqpnaClient(protocol="auto") (default; env MAQPNA_MCP_PROTOCOL) probes server/discover and speaks the stateless MCP 2026-07-28 revision when the server offers it, otherwise the legacy initialize handshake. Pending approvals then arrive as input_required results, which wait_for_approval handles transparently (it echoes requestState).

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

The W3C traceparent/tracestate is sent as HTTP headers and in params._meta; the gateway links its spans and the audit record (ext.traceId) to it.

Exceptions

Exception	When
PolicyDenied(rule, reason)	JSON-RPC -32001: denied by a rule, by default-deny, or by an approver
ApprovalPending(approval_id)	JSON-RPC -32002: queued for a human (async mode only)
Unauthorized	JSON-RPC -32003, or HTTP 401/403 that persists after the token is re-read and the call retried
ApprovalTimeout	wait_for_approval timed out
InputRequired(input_requests, request_state)	MCP 2026-07-28 input_required result from an upstream (retry with input_responses=, request_state=)
HeaderMismatch / UnsupportedProtocolVersion	JSON-RPC -32020 / -32022 (MCP 2026-07-28)
RPCError(code, message, data)	Any other JSON-RPC error
TransportError	The gateway is unreachable or returned a non JSON-RPC response
ConfigError	MAQPNA_GATEWAY_URL is missing or the token is unreadable
InvalidRequest (also a ValueError)	report_result: bad result, oversized summary, or HTTP 400/413

All of them inherit from MaqpnaError.

Development

cd sdk/python
uv venv .venv && uv pip install --python .venv/bin/python -e '.[test]'   # or: python3 -m venv .venv && .venv/bin/pip install -e '.[test]'
.venv/bin/python -m pytest -q

The tests run the SDK against an in-memory MCP gateway built on httpx.MockTransport. They need no network access.

Licence

Proprietary software of Kosa8 Labs, owners of MAQPNA. See LICENSE; for licensing, write to hello@maqpna.com.