Gate client (HTTP/MCP)

This is the third door: HttpMcpTransport speaks POST /mcp, the tool catalog an MCP host sees. The gate's dg.* verbs — an agent's KnowledgeClient and the operator's CurationClient over the Unix socket or POST /rpc — are on the gate clients page. All three raise one GateError (doublegate_sdk.errors).

doublegate_sdk.client connects an application or agent to a running gate over the product's public integration surface: POST /mcp (ADR-0050). It is an explicit, optional import using only the Python standard library, and it adds no connection, endpoint discovery or credential discovery on import or on construction.

It is a transport, not an authority. A read is not an admission and a proposal is not an approval.

The narrowed contract — read this first

The client speaks the tools the maintained client tier actually serves, and nothing else. Two absences matter:

Not available	Why
`inventory` / `inventory_pages`	The client tier serves no inventory tool. Rather than ship a method that could only ever fail, there is none.
Whole-gate `status()`	`doublegate.status` requires an artifact id. There is no gate-wide status call on this surface.

The served catalog is doublegate.remember, recall, status, why, tip, pending, rank, hide, ban. Ask a specific server what it serves rather than assuming:

transport.discover()    # server/discover: protocol versions, serverInfo, capabilities
transport.tool_names()  # tools/list: the tool names this server actually lists

A tool outside the SDK's declared catalog raises ValueError locally instead of being sent and blamed on the server.

Connect

import os
from doublegate_sdk import connect

client = connect('https://gate.example.org/mcp', token=os.environ['GATE_TOKEN'])

record = client.status('sha256:' + '…')
print(record['state'])

answers = client.recall('basalt specimens', limit=5)
for hit in answers['results']:
    print(hit['artifact_id'])

connect() is the one call most callers need. It builds an HttpMcpTransport and wraps it in a GateClient, and takes the same bounds and credential arguments; assemble the two by hand only when you want to hold the transport itself, for discover() or tool_names():

from doublegate_sdk.client import GateClient, HttpMcpTransport

transport = HttpMcpTransport('https://gate.example.org/mcp', token=token)
client = GateClient(transport)
print(transport.tool_names())

connect() also exposes the transport as client.transport, so the low-level form is rarely necessary.

HTTPS is the default and the only option for a remote endpoint. Plain HTTP is refused unless you pass allow_insecure_loopback=True and the host is loopback — that combination exists for tests and for a daemon on this host, and it still refuses a remote host:

connect('http://127.0.0.1:8480/mcp', allow_insecure_loopback=True)

The endpoint is always explicit. The SDK reads no configuration file, no environment variable and no well-known path to find a gate.

Propose

Writes are off unless you turn them on. The transport itself refuses a write-annotated tool, so a read-only client cannot be talked into writing by a caller that reaches past GateClient:

writer = connect(endpoint, token=token, allow_writes=True)
proposal = writer.propose(
    'The laboratory labels basalt specimens by collection date.',
    content_type='memory', trust_class='T-4', source_uri='application://observation/123',
)
state = writer.status(proposal['artifact_id'])['state']

content is text. Bytes are accepted only when they are valid UTF-8, and are decoded; binary is refused rather than base64-smuggled into a field the service would store as literal base64 text. The service's remember schema has no encoding parameter, so there is nothing to smuggle it into honestly.

Writer identity is never sent. The service derives identity from the connection and refuses a request carrying writer_identity or deployment_id.

recall excludes the caller's own unreviewed pending echoes, so an application does not read back its own un-admitted proposal and mistake it for an answer. Provisional knowledge is excluded by default; pass include_provisional=True when your policy permits it. Each hit keeps its own provenance flags, and no score is relabelled as trust or confidence.

Bounds and errors

timeout (default 10s) is one total deadline for the whole exchange, not a per-socket operation timeout: it covers connect, send, the status line, the response headers and the body, and is re-checked before every read, so a peer cannot extend it by answering one byte at a time. It is enforced on the calling thread — no background worker is started, so nothing can outlive the call.

One thing the deadline does not cover: name resolution. The standard library resolves the host inside connect() with no cancellation hook, so a hanging DNS server is bounded by your resolver's own timeout, not by this. Pass an IP literal if that matters to you.

Responses are capped at 1 MiB (max_response_bytes) and requests at 1 MiB (max_request_bytes) — separate budgets, so lowering one does not silently trip the other. A 200 whose Content-Type is not application/json (or a +json type) is refused as invalid_response, even when its body would have parsed: a captive portal or proxy interstitial is not this gate's answer.

The endpoint itself is validated at construction, so a local configuration mistake is a ValueError you get immediately rather than a GateError that looks like the gate being down:

credentials in the URL are refused. https://user:[email protected]/mcp raises — urlsplit strips userinfo, so such a credential is never sent, and keeping it would only leave it in transport.endpoint waiting to be logged. Pass token= instead. An empty username (https://@gate.example/mcp) is still userinfo and is still refused.
control characters and whitespace in the path are refused, as are tab, CR and LF anywhere in the URL — urlsplit deletes those silently, which would otherwise turn /mcp\tx into /mcpx behind your back.
a token the HTTP header encoding cannot carry is refused at construction. Headers are Latin-1; a token with CJK, emoji or Cyrillic raises ValueError when you build the transport rather than UnicodeEncodeError mid-request.

Nothing is retried. A write repeated without an idempotency contract is worse than a write that failed loudly. Redirects are refused, never followed: following one would carry the bearer token to whatever host the response names.

GateError.kind is a stable string; branch on it rather than parsing English.

`kind`	Meaning
`unavailable`	The endpoint could not be reached.
`timeout`	The total deadline expired (connect, send, headers or body).
`unauthorized` / `forbidden`	HTTP 401 / 403 from the serving surface.
`redirect_refused`	The server answered a redirect; it was not followed.
`unsupported_operation`	JSON-RPC `-32601`/`-32602`, or HTTP 405 — this build does not serve that.
`remote_error`	The server refused; the numeric `code` is kept.
`tool_error`	The tool answered with `isError`.
`invalid_response`	The response did not satisfy the checked shape.
`request_too_large` / `response_too_large`	A byte budget was exceeded.
`writes_disabled`	A write tool was called without `allow_writes=True`.

Remote error text is withheld deliberately; the numeric code is kept. Server messages carry internals, and a client log is the wrong place for them. This version does not claim a universal authorization taxonomy — map server codes explicitly in your application.

GateError.outcome_unknown is true when a write may have been received but the outcome could not be read (timeout after sending, a truncated or malformed response, a server -32603). Reconcile before resubmitting; the service's content-duplicate detection is not an exactly-once or idempotency-key guarantee.

What the MCP support actually is

The server serves the 2026-07-28 stateless JSON request/response subset of Streamable HTTP, and so does this client:

one JSON-RPC message per POST /mcp, one JSON object back;
no SSE, no streaming — GET /mcp is a 405 by design;
no sessions — the server issues no Mcp-Session-Id and the client sends none;
negotiation is server/discover and tools/list. initialize is refused by the server unless an operator enables a legacy flag, so the SDK does not send it.

This is not a general MCP framework and does not claim to be one. describe_client()['mcp'] reports these facts, and a test asserts the SDK never claims streaming or session support.

Explicit transport injection

GateClient accepts any McpTransport implementing call(tool, arguments) -> dict. A host application can supply its own supported transport — an MCP client it already runs, for instance — without the SDK importing an agent framework. A custom injected transport owns its own write-enablement contract; none of these client-side flags grant server authority.

Verified scope

Proved by the scripted-responder suite (tests/test_memory_client.py, real loopback HTTP servers, real bytes — not mock transport objects): tool mapping, bearer header presence and absence, token never appearing in an error, response and request bounds as separate budgets, redirect refusal, 401/403/405 mapping, JSON-RPC error-code mapping, write opt-in at both client and transport layers, UTF-8 enforcement, unknown-outcome flagging, and the absence of an inventory method. The suite also asserts that no socket transport is re-exported, so a removed transport cannot quietly return.

Proved by one isolated real-service run (examples/verify_memory_lifecycle.py, against the product's own console POST /mcp on an ephemeral loopback port): server/discover and tools/list answered by the real server, a propose → status → recall cycle with the record's state coming from the service, own pending excluded from recall, provisional requiring opt-in, a read-only client refusing a write, and the served catalog containing no inventory tool.

Not proved, and why

The MCP endpoint on the maintained client tier is keyless

console.py dispatches POST /mcp before its _authed() check; only /api/* routes enforce the console bearer token. On that tier, loopback is the credential (INV-SEC-11 / FR-65).

So the real-service run proves the functional path over MCP, and proves the product's actual token guard separately on /api/status (401 without a token, 200 with). It does not prove an authenticated remote MCP client. The SDK's own 401/403 handling is proved against scripted responders. No server-side change was made to close this; it is the service's to close.

Also not covered: HTTPS against a real certificate chain, a remote (non-loopback) deployment, an org-tier keyed MCP surface, cross-tenant authorization, installed-wheel or Windows execution, and organization delivery. This is source-mode verification on Linux/WSL.

Generated reference

doublegate_sdk.client

Explicit gate client over the product's public HTTP/MCP endpoint.

The gate's supported network surface is POST /mcp (ADR-0050): one JSON-RPC message in, one JSON object out. :class:HttpMcpTransport speaks that and only that. It is not an authorization layer, not a general MCP framework, and not a streaming client — the server issues no session id and offers no SSE stream, so neither does this.

Nothing connects on import or on construction. There is no endpoint discovery, no credential discovery, no redirect following and no automatic retry.

The tool catalog below is the maintained client tier's, read from the service's own schema. It is deliberately narrow: there is no inventory tool, and status requires an artifact id. See docs/api/client.md.

GateError

Bases: DoublegateError, RuntimeError

A fixed diagnostic. str(exc) is the kind (and code); the server's text is detail.

Source code in src/doublegate_sdk/errors.py

class GateError(DoublegateError, RuntimeError):
    """A fixed diagnostic. ``str(exc)`` is the kind (and code); the server's text is ``detail``."""

    def __init__(self, kind: str, code: int | None = None, *, retry_after_ms: int | None = None,
                 outcome_unknown: bool = False, detail: str | None = None):
        if kind not in KINDS:
            raise ValueError(f'unknown error kind {kind!r}')
        self.kind, self.code = kind, code
        self.retry_after_ms = retry_after_ms
        self.outcome_unknown = outcome_unknown
        self.detail = None if detail is None else str(detail)[:DETAIL_MAX]
        super().__init__(kind if code is None else f'{kind} ({code})')

    def __repr__(self) -> str:
        return f'GateError(kind={self.kind!r}, code={self.code!r}, outcome_unknown={self.outcome_unknown!r})'

retry_after_ms `instance-attribute`

retry_after_ms = retry_after_ms

outcome_unknown `instance-attribute`

outcome_unknown = outcome_unknown

detail `instance-attribute`

detail = None if detail is None else str(detail)[:DETAIL_MAX]

McpTransport

Bases: Protocol

Explicit tool transport over POST /mcp; the receiving gate remains authoritative.

Distinct from doublegate_sdk.transport.GateTransport, whose call takes a dg.* verb for POST /rpc or the Unix socket. Same shape, different vocabulary.

Source code in src/doublegate_sdk/client.py

class McpTransport(Protocol):
    """Explicit tool transport over ``POST /mcp``; the receiving gate remains authoritative.

    Distinct from ``doublegate_sdk.transport.GateTransport``, whose ``call`` takes a
    ``dg.*`` verb for ``POST /rpc`` or the Unix socket. Same shape, different vocabulary.
    """

    def call(self, tool: str, arguments: dict[str, Any]) -> dict[str, Any]: ...

call

call(tool: str, arguments: dict[str, Any]) -> dict[str, Any]

Source code in src/doublegate_sdk/client.py

def call(self, tool: str, arguments: dict[str, Any]) -> dict[str, Any]: ...

HttpMcpTransport

The product's POST /mcp endpoint, bounded and explicit.

HTTPS is the default and the only remote option. allow_insecure_loopback exists for tests and for a daemon on this host; it refuses a remote host.

Writes are off unless allow_writes=True — the transport refuses a write-annotated tool itself, so a read-only client cannot be talked into writing by a caller that reaches past :class:GateClient.

timeout is a total deadline for the whole exchange, not a per-socket operation timeout: it covers connect, send, the status line, the headers and the body, and is re-checked before every read, so a peer cannot extend it by answering slowly. It is enforced on the calling thread with no background worker. One thing it does not cover: name resolution, which the standard library performs inside connect() with no cancellation hook — a DNS server that hangs is bounded by the resolver, not by this. Use an IP literal or a resolver timeout if that matters to you.

Nothing is retried: a write repeated without an idempotency contract is worse than a write that failed loudly.

Source code in src/doublegate_sdk/client.py

class HttpMcpTransport:
    """The product's ``POST /mcp`` endpoint, bounded and explicit.

    HTTPS is the default and the only remote option. ``allow_insecure_loopback``
    exists for tests and for a daemon on this host; it refuses a remote host.

    Writes are off unless ``allow_writes=True`` — the transport refuses a
    write-annotated tool itself, so a read-only client cannot be talked into
    writing by a caller that reaches past :class:`GateClient`.

    ``timeout`` is a total deadline for the whole exchange, not a per-socket
    operation timeout: it covers connect, send, the status line, the headers and
    the body, and is re-checked before every read, so a peer cannot extend it by
    answering slowly. It is enforced on the calling thread with no background
    worker. One thing it does **not** cover: name resolution, which the standard
    library performs inside ``connect()`` with no cancellation hook — a DNS
    server that hangs is bounded by the resolver, not by this. Use an IP literal
    or a resolver timeout if that matters to you.

    Nothing is retried: a write repeated without an idempotency contract is
    worse than a write that failed loudly.
    """

    #: The stateless JSON request/response subset the server actually serves.
    #: No ``Mcp-Session-Id``, no SSE, no ``initialize`` handshake by default.
    PROTOCOL_SUBSET = 'stateless-json-request-response'

    def __init__(self, endpoint: str, *, token: str | None = None, timeout: float = 10.0,
                 max_response_bytes: int = 1_048_576, max_request_bytes: int = 1_048_576,
                 allow_writes: bool = False, allow_insecure_loopback: bool = False,
                 ssl_context: ssl.SSLContext | None = None):
        if type(allow_insecure_loopback) is not bool:
            raise ValueError('allow_insecure_loopback must be boolean')
        if type(allow_writes) is not bool:
            raise ValueError('allow_writes must be boolean')
        if not _positive_finite(timeout):
            raise ValueError('timeout must be finite and positive')
        if not _integer(max_response_bytes, 1):
            raise ValueError('max_response_bytes must be a positive integer')
        if not _integer(max_request_bytes, 1):
            raise ValueError('max_request_bytes must be a positive integer')
        if token is not None:
            if (not isinstance(token, str) or not token
                    or any(c in token for c in '\r\n')):
                raise ValueError('token must be nonempty single-line text')
            try:
                # ``http.client`` encodes headers as Latin-1. A token outside
                # that range would otherwise blow up inside ``request()`` as a
                # raw UnicodeEncodeError, mid-call and outside the taxonomy.
                # A credential problem belongs at construction.
                token.encode('latin-1')
            except UnicodeEncodeError:
                raise ValueError('token must be nonempty single-line text') from None
        self._target = _parse_endpoint(endpoint, allow_insecure_loopback=allow_insecure_loopback)
        self.endpoint = endpoint
        self._token = token
        self._timeout = float(timeout)
        self.max_response_bytes = max_response_bytes
        self.max_request_bytes = max_request_bytes
        self._allow_writes = allow_writes
        self._ssl_context = ssl_context

    # ---- protocol negotiation (only what the server supports) ----

    def discover(self) -> dict[str, Any]:
        """``server/discover`` — protocol versions, server info, capabilities.

        This is the negotiation method the maintained server always answers.
        ``initialize`` is refused unless the operator turned on a legacy flag,
        so the SDK does not send it.
        """
        return self._rpc('server/discover', {}, writing=False)

    def tool_names(self) -> tuple[str, ...]:
        """The tools this server actually lists, in its own order."""
        result = self._rpc('tools/list', {}, writing=False)
        tools = result.get('tools')
        _require(isinstance(tools, list))
        names = []
        for tool in tools:
            _require(isinstance(tool, dict) and isinstance(tool.get('name'), str))
            names.append(tool['name'])
        return tuple(names)

    # ---- tool calls ----

    def call(self, tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
        if tool not in _TOOLS:
            raise ValueError(f'unknown tool for this client tier: {tool!r}')
        writing = _TOOLS[tool]
        if writing and not self._allow_writes:
            raise GateError('writes_disabled')
        result = self._rpc('tools/call', {'name': tool, 'arguments': arguments}, writing=writing)
        if result.get('isError'):
            raise GateError('tool_error', outcome_unknown=writing)
        payload = result.get('structuredContent')
        if not isinstance(payload, dict):
            # The text block carries the same payload, but a response missing
            # the structured half is not one this client will guess at.
            raise GateError('invalid_response', outcome_unknown=writing)
        return payload

    # ---- the wire ----

    def _rpc(self, method: str, params: dict[str, Any], *, writing: bool) -> dict[str, Any]:
        try:
            body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method,
                               'params': params}, allow_nan=False).encode()
        except (ValueError, TypeError):
            raise ValueError('arguments must be JSON-serializable without NaN or Infinity') from None
        if len(body) > self.max_request_bytes:
            raise GateError('request_too_large')

        headers = {'Content-Type': 'application/json', 'Accept': 'application/json',
                   'Content-Length': str(len(body))}
        if self._token is not None:
            headers['Authorization'] = f'Bearer {self._token}'

        deadline = time.monotonic() + self._timeout
        sending = False
        connection = None
        try:
            connection = self._connect(deadline)
            # ``sending`` stays conservative: it is set before the socket is
            # touched, so a failure anywhere from connect onwards reports a
            # write's outcome as unknown rather than guessing it did not land.
            sending = True
            connection.connect()          # explicit: ``sock`` must exist to arm
            connection.sock.settimeout(self._remaining(deadline))
            connection.request('POST', self._target.path, body=body, headers=headers)
            # Read the status line, the headers and the body through one
            # deadline-bounded file. A slow drip of *headers* is bounded by the
            # same budget as a slow drip of body bytes.
            connection.sock = _DeadlineSocket(connection.sock,
                                              lambda: self._remaining(deadline))
            response = connection.getresponse()
            status = response.status
            content_type = response.getheader('Content-Type')
            if status in (301, 302, 303, 307, 308):
                # Following a redirect would carry the bearer token to whatever
                # host the answer names. It is refused, not followed.
                raise GateError('redirect_refused', status)
            raw = self._read_bounded(response, deadline)
            if len(raw) > self.max_response_bytes:
                raise GateError('response_too_large', outcome_unknown=writing)
        except GateError:
            raise
        except TimeoutError:
            raise GateError('timeout', outcome_unknown=writing and sending) from None
        except (OSError, http.client.HTTPException):
            raise GateError('unavailable', outcome_unknown=writing and sending) from None
        finally:
            if connection is not None:
                connection.close()

        if status == 401:
            raise GateError('unauthorized', 401)
        if status == 403:
            raise GateError('forbidden', 403)
        if status == 413:
            raise GateError('request_too_large', 413)
        if status == 405:
            raise GateError('unsupported_operation', 405)
        if status == 202:
            # A notification was accepted. This client sends none.
            raise GateError('invalid_response', 202, outcome_unknown=writing)
        if status != 200:
            raise GateError('remote_error', status, outcome_unknown=writing)
        if not _is_json_media_type(content_type):
            # The request said ``Accept: application/json``. A 200 labelled
            # anything else is an interstitial or a proxy, not this gate's
            # answer — even when its body happens to parse.
            raise GateError('invalid_response', outcome_unknown=writing)
        return self._parse(raw, writing=writing)

    def _read_bounded(self, response: Any, deadline: float) -> bytes:
        """Read at most ``max_response_bytes + 1`` bytes, deadline enforced.

        ``read(n)`` would loop inside the buffered reader until it had all ``n``
        bytes; ``read1`` returns what one recv produced, so the budget is checked
        between chunks as well as before each recv.
        """
        limit = self.max_response_bytes + 1
        chunks: list[bytes] = []
        total = 0
        while total < limit:
            self._remaining(deadline)
            chunk = response.read1(limit - total)
            if not chunk:
                break
            chunks.append(chunk)
            total += len(chunk)
        return b''.join(chunks)

    def _connect(self, deadline: float) -> Any:
        timeout = self._remaining(deadline)
        if self._target.scheme == 'https':
            context = self._ssl_context or ssl.create_default_context()
            return http.client.HTTPSConnection(self._target.host, self._target.port,
                                               timeout=timeout, context=context)
        return http.client.HTTPConnection(self._target.host, self._target.port, timeout=timeout)

    def _remaining(self, deadline: float) -> float:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError
        return remaining

    def _parse(self, raw: bytes, *, writing: bool) -> dict[str, Any]:
        try:
            message = json.loads(raw, object_pairs_hook=_unique_object)
            _require(isinstance(message, dict) and message.get('jsonrpc') == '2.0')
            _require(type(message.get('id')) is int and message['id'] == 1)
            _require(('result' in message) != ('error' in message))
            if 'error' in message:
                code = message['error']['code']
                _require(type(code) is int)
                kind = 'unsupported_operation' if code in _UNSUPPORTED_CODES else 'remote_error'
                # -32603 is the server's own internal failure: a write may or
                # may not have landed, and saying otherwise would be a guess.
                raise GateError(kind, code, outcome_unknown=writing and code == -32603)
            _require(isinstance(message['result'], dict))
            return message['result']
        except GateError as error:
            if error.kind == 'invalid_response' and writing:
                error.outcome_unknown = True
            raise
        except (ValueError, KeyError, TypeError, RecursionError):
            # RecursionError: ``json.loads`` on a deeply nested document, well
            # inside the byte bound. ``package.py`` already treats it as bad
            # input; a caller's ``except DoublegateError`` must catch it here
            # too, and a write must still report its outcome as unknown.
            raise GateError('invalid_response', outcome_unknown=writing) from None

PROTOCOL_SUBSET `class-attribute` `instance-attribute`

PROTOCOL_SUBSET = 'stateless-json-request-response'

endpoint `instance-attribute`

endpoint = endpoint

max_response_bytes `instance-attribute`

max_response_bytes = max_response_bytes

max_request_bytes `instance-attribute`

max_request_bytes = max_request_bytes

discover

discover() -> dict[str, Any]

server/discover — protocol versions, server info, capabilities.

This is the negotiation method the maintained server always answers. initialize is refused unless the operator turned on a legacy flag, so the SDK does not send it.

Source code in src/doublegate_sdk/client.py

def discover(self) -> dict[str, Any]:
    """``server/discover`` — protocol versions, server info, capabilities.

    This is the negotiation method the maintained server always answers.
    ``initialize`` is refused unless the operator turned on a legacy flag,
    so the SDK does not send it.
    """
    return self._rpc('server/discover', {}, writing=False)

tool_names

tool_names() -> tuple[str, ...]

The tools this server actually lists, in its own order.

Source code in src/doublegate_sdk/client.py

def tool_names(self) -> tuple[str, ...]:
    """The tools this server actually lists, in its own order."""
    result = self._rpc('tools/list', {}, writing=False)
    tools = result.get('tools')
    _require(isinstance(tools, list))
    names = []
    for tool in tools:
        _require(isinstance(tool, dict) and isinstance(tool.get('name'), str))
        names.append(tool['name'])
    return tuple(names)

call

call(tool: str, arguments: dict[str, Any]) -> dict[str, Any]

Source code in src/doublegate_sdk/client.py

def call(self, tool: str, arguments: dict[str, Any]) -> dict[str, Any]:
    if tool not in _TOOLS:
        raise ValueError(f'unknown tool for this client tier: {tool!r}')
    writing = _TOOLS[tool]
    if writing and not self._allow_writes:
        raise GateError('writes_disabled')
    result = self._rpc('tools/call', {'name': tool, 'arguments': arguments}, writing=writing)
    if result.get('isError'):
        raise GateError('tool_error', outcome_unknown=writing)
    payload = result.get('structuredContent')
    if not isinstance(payload, dict):
        # The text block carries the same payload, but a response missing
        # the structured half is not one this client will guess at.
        raise GateError('invalid_response', outcome_unknown=writing)
    return payload

GateClient

Propose observations and read gate answers through an explicit transport.

Usually built by :func:connect. Presence in a read is not admission, and a proposal is not an approval. The gate decides; this object carries bytes.

Source code in src/doublegate_sdk/client.py

class GateClient:
    """Propose observations and read gate answers through an explicit transport.

    Usually built by :func:`connect`. Presence in a read is not admission, and a
    proposal is not an approval. The gate decides; this object carries bytes.
    """

    def __init__(self, transport: McpTransport):
        self._transport = transport

    @property
    def transport(self) -> McpTransport:
        """The underlying transport — for ``discover()`` / ``tool_names()``."""
        return self._transport

    @staticmethod
    def _artifact_id(artifact_id: Any) -> str:
        if not isinstance(artifact_id, str) or not artifact_id.strip():
            raise ValueError('artifact_id must be a nonempty string')
        return artifact_id

    def status(self, artifact_id: str) -> dict[str, Any]:
        """One artifact's position in the register.

        The maintained tool requires an artifact id; there is no whole-gate
        status call on this surface.
        """
        result = self._transport.call('doublegate.status',
                                      {'artifact_id': self._artifact_id(artifact_id)})
        if not isinstance(result, dict) or not isinstance(result.get('state'), str):
            raise GateError('invalid_response')
        return result

    def why(self, artifact_id: str) -> dict[str, Any]:
        """Every ledger event that moved one artifact, in order."""
        result = self._transport.call('doublegate.why',
                                      {'artifact_id': self._artifact_id(artifact_id)})
        if not isinstance(result, dict) or not isinstance(result.get('events'), list):
            raise GateError('invalid_response')
        return result

    def propose(self, content: str | bytes, *, content_type: str, source_uri: str,
                trust_class: str | None = None, space: str | None = None,
                derives_from: list[str] | None = None) -> dict[str, Any]:
        """Submit content for admission. Not an approval, and never retried.

        ``content`` is text. Bytes are accepted only when they are valid UTF-8
        and are decoded; binary is refused rather than base64-smuggled into a
        text field the service would store verbatim.

        Writer identity is never sent: the service derives it from the
        connection and refuses a request that carries it.
        """
        if isinstance(content, bytes):
            try:
                content = content.decode('utf-8')
            except UnicodeDecodeError:
                raise ValueError('content must be UTF-8 text; this surface stores no binary') from None
        if not isinstance(content, str) or not content:
            raise ValueError('content must be nonempty text')
        if any(not isinstance(v, str) or not v for v in (content_type, source_uri)):
            raise ValueError('content_type and source_uri are required nonempty strings')
        arguments: dict[str, Any] = {'content': content, 'content_type': content_type,
                                     'source_uri': source_uri}
        for name, value in (('trust_class', trust_class), ('space', space)):
            if value is not None:
                if not isinstance(value, str) or not value:
                    raise ValueError(f'{name} must be nonempty text when given')
                arguments[name] = value
        if derives_from is not None:
            if (not isinstance(derives_from, list)
                    or any(not isinstance(p, str) or not p for p in derives_from)):
                raise ValueError('derives_from must be a list of nonempty artifact ids')
            arguments['derives_from'] = list(derives_from)
        result = self._transport.call('doublegate.remember', arguments)
        if (not isinstance(result, dict) or not isinstance(result.get('artifact_id'), str)
                or not result['artifact_id'] or not isinstance(result.get('state'), str)
                or not result['state']):
            raise GateError('invalid_response', outcome_unknown=True)
        return result

    def recall(self, query: str, *, limit: int = 10, spaces: list[str] | None = None,
               include_provisional: bool = False, audience: str | None = None) -> dict[str, Any]:
        """Read served knowledge.

        Own unreviewed echoes are excluded by default, so a caller does not read
        back its own un-admitted proposal and mistake it for an answer.
        """
        if not isinstance(query, str) or not query.strip():
            raise ValueError('query must be nonempty text')
        if not _integer(limit, 1):
            raise ValueError('limit must be a positive integer')
        if type(include_provisional) is not bool:
            raise ValueError('include_provisional must be boolean')
        if spaces is not None and (not isinstance(spaces, list) or not spaces
                                   or any(not isinstance(s, str) or not s for s in spaces)):
            raise ValueError('spaces must be a nonempty list of strings')
        arguments: dict[str, Any] = {'query': query, 'k': limit, 'include_own_pending': False,
                                     'include_provisional': include_provisional}
        if spaces is not None:
            arguments['spaces'] = spaces
        if audience is not None:
            if not isinstance(audience, str) or not audience:
                raise ValueError('audience must be nonempty text when given')
            arguments['audience'] = audience
        result = self._transport.call('doublegate.recall', arguments)
        if (not isinstance(result, dict) or not isinstance(result.get('results'), list)
                or not all(isinstance(hit, dict) for hit in result['results'])
                or len(result['results']) > limit):
            raise GateError('invalid_response')
        return result

    def pending(self, *, limit: int = 100) -> dict[str, Any]:
        """What is waiting for review, as metadata only — never content."""
        if not _integer(limit, 1):
            raise ValueError('limit must be a positive integer')
        result = self._transport.call('doublegate.pending', {'limit': limit})
        if (not isinstance(result, dict) or not isinstance(result.get('pending'), list)
                or not all(isinstance(item, dict) for item in result['pending'])
                or len(result['pending']) > limit):
            raise GateError('invalid_response')
        return result

transport `property`

transport: McpTransport

The underlying transport — for discover() / tool_names().

status

status(artifact_id: str) -> dict[str, Any]

One artifact's position in the register.

The maintained tool requires an artifact id; there is no whole-gate status call on this surface.

Source code in src/doublegate_sdk/client.py

def status(self, artifact_id: str) -> dict[str, Any]:
    """One artifact's position in the register.

    The maintained tool requires an artifact id; there is no whole-gate
    status call on this surface.
    """
    result = self._transport.call('doublegate.status',
                                  {'artifact_id': self._artifact_id(artifact_id)})
    if not isinstance(result, dict) or not isinstance(result.get('state'), str):
        raise GateError('invalid_response')
    return result

why

why(artifact_id: str) -> dict[str, Any]

Every ledger event that moved one artifact, in order.

Source code in src/doublegate_sdk/client.py

def why(self, artifact_id: str) -> dict[str, Any]:
    """Every ledger event that moved one artifact, in order."""
    result = self._transport.call('doublegate.why',
                                  {'artifact_id': self._artifact_id(artifact_id)})
    if not isinstance(result, dict) or not isinstance(result.get('events'), list):
        raise GateError('invalid_response')
    return result

propose

propose(content: str | bytes, *, content_type: str, source_uri: str, trust_class: str | None = None, space: str | None = None, derives_from: list[str] | None = None) -> dict[str, Any]

Submit content for admission. Not an approval, and never retried.

content is text. Bytes are accepted only when they are valid UTF-8 and are decoded; binary is refused rather than base64-smuggled into a text field the service would store verbatim.

Writer identity is never sent: the service derives it from the connection and refuses a request that carries it.

Source code in src/doublegate_sdk/client.py

def propose(self, content: str | bytes, *, content_type: str, source_uri: str,
            trust_class: str | None = None, space: str | None = None,
            derives_from: list[str] | None = None) -> dict[str, Any]:
    """Submit content for admission. Not an approval, and never retried.

    ``content`` is text. Bytes are accepted only when they are valid UTF-8
    and are decoded; binary is refused rather than base64-smuggled into a
    text field the service would store verbatim.

    Writer identity is never sent: the service derives it from the
    connection and refuses a request that carries it.
    """
    if isinstance(content, bytes):
        try:
            content = content.decode('utf-8')
        except UnicodeDecodeError:
            raise ValueError('content must be UTF-8 text; this surface stores no binary') from None
    if not isinstance(content, str) or not content:
        raise ValueError('content must be nonempty text')
    if any(not isinstance(v, str) or not v for v in (content_type, source_uri)):
        raise ValueError('content_type and source_uri are required nonempty strings')
    arguments: dict[str, Any] = {'content': content, 'content_type': content_type,
                                 'source_uri': source_uri}
    for name, value in (('trust_class', trust_class), ('space', space)):
        if value is not None:
            if not isinstance(value, str) or not value:
                raise ValueError(f'{name} must be nonempty text when given')
            arguments[name] = value
    if derives_from is not None:
        if (not isinstance(derives_from, list)
                or any(not isinstance(p, str) or not p for p in derives_from)):
            raise ValueError('derives_from must be a list of nonempty artifact ids')
        arguments['derives_from'] = list(derives_from)
    result = self._transport.call('doublegate.remember', arguments)
    if (not isinstance(result, dict) or not isinstance(result.get('artifact_id'), str)
            or not result['artifact_id'] or not isinstance(result.get('state'), str)
            or not result['state']):
        raise GateError('invalid_response', outcome_unknown=True)
    return result

recall

recall(query: str, *, limit: int = 10, spaces: list[str] | None = None, include_provisional: bool = False, audience: str | None = None) -> dict[str, Any]

Read served knowledge.

Own unreviewed echoes are excluded by default, so a caller does not read back its own un-admitted proposal and mistake it for an answer.

Source code in src/doublegate_sdk/client.py

def recall(self, query: str, *, limit: int = 10, spaces: list[str] | None = None,
           include_provisional: bool = False, audience: str | None = None) -> dict[str, Any]:
    """Read served knowledge.

    Own unreviewed echoes are excluded by default, so a caller does not read
    back its own un-admitted proposal and mistake it for an answer.
    """
    if not isinstance(query, str) or not query.strip():
        raise ValueError('query must be nonempty text')
    if not _integer(limit, 1):
        raise ValueError('limit must be a positive integer')
    if type(include_provisional) is not bool:
        raise ValueError('include_provisional must be boolean')
    if spaces is not None and (not isinstance(spaces, list) or not spaces
                               or any(not isinstance(s, str) or not s for s in spaces)):
        raise ValueError('spaces must be a nonempty list of strings')
    arguments: dict[str, Any] = {'query': query, 'k': limit, 'include_own_pending': False,
                                 'include_provisional': include_provisional}
    if spaces is not None:
        arguments['spaces'] = spaces
    if audience is not None:
        if not isinstance(audience, str) or not audience:
            raise ValueError('audience must be nonempty text when given')
        arguments['audience'] = audience
    result = self._transport.call('doublegate.recall', arguments)
    if (not isinstance(result, dict) or not isinstance(result.get('results'), list)
            or not all(isinstance(hit, dict) for hit in result['results'])
            or len(result['results']) > limit):
        raise GateError('invalid_response')
    return result

pending

pending(*, limit: int = 100) -> dict[str, Any]

What is waiting for review, as metadata only — never content.

Source code in src/doublegate_sdk/client.py

def pending(self, *, limit: int = 100) -> dict[str, Any]:
    """What is waiting for review, as metadata only — never content."""
    if not _integer(limit, 1):
        raise ValueError('limit must be a positive integer')
    result = self._transport.call('doublegate.pending', {'limit': limit})
    if (not isinstance(result, dict) or not isinstance(result.get('pending'), list)
            or not all(isinstance(item, dict) for item in result['pending'])
            or len(result['pending']) > limit):
        raise GateError('invalid_response')
    return result

connect

connect(endpoint: str, *, token: str | None = None, allow_writes: bool = False, timeout: float = 10.0, allow_insecure_loopback: bool = False, max_response_bytes: int = 1048576, max_request_bytes: int = 1048576, ssl_context: SSLContext | None = None) -> 'GateClient'

Build a client for one gate endpoint. The one call most callers need.

connect('https://gate.example.org/mcp', token=...) is equivalent to wrapping :class:HttpMcpTransport in a :class:GateClient, and exists so a caller does not have to assemble a transport to make an ordinary request.

It connects to nothing: the first request happens on the first method call. Every bound and every credential is still explicit, and writes remain off unless allow_writes=True.

Source code in src/doublegate_sdk/client.py

def connect(endpoint: str, *, token: str | None = None, allow_writes: bool = False,
            timeout: float = 10.0, allow_insecure_loopback: bool = False,
            max_response_bytes: int = 1_048_576, max_request_bytes: int = 1_048_576,
            ssl_context: ssl.SSLContext | None = None) -> 'GateClient':
    """Build a client for one gate endpoint. The one call most callers need.

    ``connect('https://gate.example.org/mcp', token=...)`` is equivalent to
    wrapping :class:`HttpMcpTransport` in a :class:`GateClient`, and exists so a
    caller does not have to assemble a transport to make an ordinary request.

    It connects to nothing: the first request happens on the first method call.
    Every bound and every credential is still explicit, and writes remain off
    unless ``allow_writes=True``.
    """
    return GateClient(HttpMcpTransport(
        endpoint, token=token, timeout=timeout, allow_writes=allow_writes,
        allow_insecure_loopback=allow_insecure_loopback,
        max_response_bytes=max_response_bytes, max_request_bytes=max_request_bytes,
        ssl_context=ssl_context))

describe_client

describe_client() -> dict[str, Any]

Describe this SDK's Python API offline. Not a remote gate's capabilities.

Source code in src/doublegate_sdk/client.py

def describe_client() -> dict[str, Any]:
    """Describe this SDK's Python API offline. Not a remote gate's capabilities."""
    operations = {}
    for name, metadata in _OPERATIONS.items():
        signature = inspect.signature(getattr(GateClient, name))
        parameters = []
        for parameter in signature.parameters.values():
            if parameter.name == 'self':
                continue
            variadic = parameter.kind in (inspect.Parameter.VAR_KEYWORD, inspect.Parameter.VAR_POSITIONAL)
            value = {'name': parameter.name, 'kind': parameter.kind.name,
                     'annotation': str(parameter.annotation),
                     'required': parameter.default is inspect.Parameter.empty and not variadic}
            if parameter.default is not inspect.Parameter.empty:
                value['default'] = parameter.default
            parameters.append(value)
        operations[name] = {**metadata, 'parameters': parameters,
                            'returns': str(signature.return_annotation)}
    return {
        'scope': 'sdk-contract-not-server-capabilities',
        'format': 'python-call-description-not-json-schema',
        'transports': ['http_mcp', 'caller_supplied'],
        'mcp': {
            'endpoint': 'POST /mcp',
            'subset': HttpMcpTransport.PROTOCOL_SUBSET,
            'streaming': False,
            'sessions': False,
            'negotiation': ['server/discover', 'tools/list'],
            'tools': sorted(_TOOLS),
        },
        'unsupported_operations': {
            'inventory': 'the maintained client tier serves no inventory tool',
            'whole_gate_status': 'doublegate.status requires an artifact id',
        },
        'proposals_are_admission': False,
        'automatic_retries': False,
        'follows_redirects': False,
        'operations': operations,
    }

Gate client (HTTP/MCP)

The narrowed contract — read this first

Connect

Propose

Bounds and errors

What the MCP support actually is

Explicit transport injection

Verified scope

Not proved, and why

Generated reference

doublegate_sdk.client

GateError

retry_after_ms instance-attribute

outcome_unknown instance-attribute

detail instance-attribute

McpTransport

call

HttpMcpTransport

PROTOCOL_SUBSET class-attribute instance-attribute

endpoint instance-attribute

max_response_bytes instance-attribute

max_request_bytes instance-attribute

discover

tool_names

call

GateClient

transport property

status

why

propose

recall

pending

connect

describe_client

retry_after_ms `instance-attribute`

outcome_unknown `instance-attribute`

detail `instance-attribute`

PROTOCOL_SUBSET `class-attribute` `instance-attribute`

endpoint `instance-attribute`

max_response_bytes `instance-attribute`

max_request_bytes `instance-attribute`

transport `property`