redis-context-gateway

Python SDK for Redis Context Gateway. One package holds both clients:

• ContextGatewayService is the control-plane client for Gateway and Upstream administration. Speakeasy generates it from the Gateway control-plane API, the same way as the other Iris SDKs, into redis_context_gateway.controlplane, which also holds its models, errors and HTTP client types.
• ContextGatewayClient is the data-plane client in redis_context_gateway.mcp. It connects agents to a Gateway's MCP endpoint with the official mcp library.

Both are exported from redis_context_gateway. Each loads on first use, so a control-plane script never imports the MCP library and an agent never imports the REST client. The data plane doesn't import the control plane at all.

Alpha, on PyPI as redis-context-gateway, generated from redislabsdev/iris@2584d193. Python 3.10 or later.

Control plane

from redis_context_gateway import ContextGatewayService

with ContextGatewayService(server_url=control_plane_url, admin_token=admin_token) as service:
    gateway = service.gateways.create(display_name="Docs", tool_exposure_mode="search")
    upstream = service.upstreams.create(
        gateway_id=gateway.id,
        display_name="Docs MCP",
        namespace="docs",
        connection={"endpoint": upstream_mcp_url},
        authentication={"methods": [{"bearer": {"value": upstream_token}}]},
    )

Models and errors come from the control plane: from redis_context_gateway.controlplane import errors, models. Every product call sends the admin token. The health probes (service.health.live(), service.health.ready()) send no credential. Async variants end in _async. See USAGE.md and docs/ for every operation and model.

A connection is exactly one variant: endpoint for an MCP Upstream or database for a Context Surface Upstream. A connection with both raises a pydantic ValidationError before any request is sent.

Context Surface data model

A Context Surface Upstream reads a Redis database through the data model you give it. redis_context_gateway.datamodel types that model with TypedDict classes that follow the iris data model, so a type checker catches a misspelled key or an unknown index type. They are plain dicts at run time. The control plane validates the model and names each refused member by its path under contextSurface.dataModel.

from redis_context_gateway.datamodel import DataModel

data_model: DataModel = {
    "title": "Healthcare scheduling",
    "description": "Clinics and the providers who work at them",
    "entities": [
        {
            "name": "Location",
            "description": "A medical facility where providers see patients",
            "redis_key_template": "location:{id}",
            "fields": [
                {"name": "id", "type": "str", "description": "Location ID", "is_key_component": True, "redis_indices": []},
                {"name": "name", "type": "str", "description": "Facility name", "redis_indices": [{"type": "text", "weight": 1.0}]},
                {"name": "city", "type": "str", "description": "City", "redis_indices": [{"type": "tag"}]},
            ],
            "relationships": [],
        },
        {
            "name": "Provider",
            "description": "A clinician who sees patients at one location",
            "redis_key_template": "provider:{id}",
            "fields": [
                {"name": "id", "type": "str", "description": "Provider ID", "is_key_component": True, "redis_indices": []},
                {"name": "specialty", "type": "str", "description": "Specialty", "redis_indices": [{"type": "tag"}]},
                {"name": "location_id", "type": "str", "description": "Home location", "redis_indices": [{"type": "tag"}]},
            ],
            "relationships": [
                {"name": "works_at", "target": "Location", "description": "The provider's home location", "source_field": "location_id"}
            ],
        },
    ],
    "entity_count": 2,
}

surface = service.upstreams.create(
    gateway_id=gateway.id,
    display_name="Scheduling",
    connection={"database": {"id": database_id}},
    context_surface={"data_model": data_model},
)

tests/context-gateway/healthcare-datamodel.json is a complete six-entity model that passes the control plane's validation. acl_field_mappings is not in the type because the control plane refuses it as not supported yet. The compact JSON of a model is at most 1 MiB.

Agents (MCP)

from redis_context_gateway import ContextGatewayClient

client = ContextGatewayClient(gateway.mcp_endpoint, agent_key)
async with client.connect() as session:
    found = await session.call_tool("search_tools", {"query": "find an issue"})
    result = await session.call_tool(
        "run_tool", {"tool": {"upstreamId": upstream.id, "originalName": "get_issue"}, "toolArgs": {"id": 42}}
    )

client.connect() opens a negotiated official mcp.ClientSession, so Lens instrumentation applies as is. Each call opens a new session. The constructor checks the URL, so an unsafe one fails before any request. Search tools, run_tool and the result handle tools are ordinary tool calls. A tool failure comes back as a result with is_error set to True, not an exception. Result fields are snake_case attributes in mcp 2, such as structured_content.

When the Gateway refuses a request with HTTP 401, 403 or 404, the client raises redis_context_gateway.mcp.GatewayHTTPError. Its status attribute is the HTTP status: 401 means the credential wasn't accepted, 403 that it has no grant for this Gateway, and 404 that no Gateway answers at the URL. mcp 2 would otherwise report these as a JSON-RPC error without the status. The error carries no response body and never the credential. It comes out of the async with client.connect() block, from connecting or from the call that got it, so catch it around the block. Any other error status is an mcp.MCPError.

from redis_context_gateway.mcp import GatewayHTTPError

try:
    async with client.connect() as session:
        result = await session.call_tool("search_tools", {"query": "find an issue"})
except GatewayHTTPError as error:
    if error.status == 401:
        ...  # get a new key or token

The credential is either of these, and both go in the agent_key argument:

• An Identity Service agent key with a context-gateway / gateway / use grant for the Gateway.
• An access token from the deployment's identity provider, when the deployment has IdP sign-in on. The Gateway still answers 403 when the signed-in caller has no access grant for it.

The client sends the value unchanged as Authorization: Bearer, the only place the Gateway accepts an access token. It doesn't sign users in or refresh tokens. To find the identity provider, read the Gateway's protected resource metadata (RFC 9728) without a credential at /.well-known/oauth-protected-resource/v1/gateways/{gatewayId}/mcp on the data-plane host, or take the resource_metadata URL from the WWW-Authenticate challenge of a 401. Its authorization_servers names the issuer and scopes_supported the scopes to request. When a token expires, build a new client with a fresh one.

The Gateway negotiates MCP protocol 2026-07-28, 2025-11-25, 2025-06-18 and 2025-03-26. The SDK requires mcp>=2.3,<3. A session asks for 2026-07-28 with server/discover, so that is the version it uses with the Gateway, and it sends that version on every later request. A server that answers server/discover with an error gets the initialize handshake instead, which offers 2025-11-25 and also accepts 2025-06-18 or 2025-03-26.

MCP reads default to a 300-second inactivity timeout, with 30-second connect, write and connection-pool budgets. Pass timeout=10 to use ten seconds for every HTTP timeout, or timeout=httpx2.Timeout(30, read=600) for a separate read budget. mcp 2 uses the httpx2 package, so the timeout and the test transport are httpx2 types. These are HTTP inactivity limits, not a total tool-call deadline. REST timeout defaults are unchanged.

Result handles

A Gateway can hold a large tool result and answer with a small handle stub instead. The agent then reads the result in pieces through seven tools:

The Gateway lists them in both exposure modes when the data plane runs with result_handles.enabled: true. Handles are held in the Gateway's database, so the Gateway also needs a databaseId. Without one, results come back whole and every handle reads as unknown.

run_tool_as_handle takes the arguments of the call it wraps. In full mode that is {"name": <tool name from tools/list>, "toolArgs": {...}}. In search mode it is {"tool": {"upstreamId": ..., "originalName": ...}, "toolArgs": {...}}, as run_tool takes. The other tools take a handle, an ID that matches ^h_[0-9a-f]{32}$.

from redis_context_gateway.mcp import HANDLE_META_KEY, PAGE_RESULTS, RELEASE_RESULTS, RUN_TOOL_AS_HANDLE

async with client.connect() as session:
    stub = await session.call_tool(
        RUN_TOOL_AS_HANDLE,
        {"tool": {"upstreamId": upstream.id, "originalName": "list_issues"}, "toolArgs": {}},
    )
    held = (stub.meta or {}).get(HANDLE_META_KEY)
    if held is not None:
        try:
            page = await session.call_tool(PAGE_RESULTS, {"handle": held["handle"], "offset": 0, "limit": 10})
        finally:
            # Release even if paging fails, or the result stays in the Gateway database until its TTL.
            await session.call_tool(RELEASE_RESULTS, {"handle": held["handle"]})

A stub carries the envelope as structuredContent and JSON text, and the same envelope without its preview and next step under _meta["com.redis.context-gateway/handle"]. A result can still come back whole, for example when the Gateway database is out of memory, so check for that key rather than assuming a stub. An unknown, expired, released or another caller's handle is a tool error result, not an exception. Handles are bound to the caller and the tenant.

redis_context_gateway.mcp exports the tool names (RUN_TOOL_AS_HANDLE, PAGE_RESULTS, QUERY_RESULTS, COUNT_RESULTS, AGGREGATE_RESULTS, DESCRIBE_RESULT, RELEASE_RESULTS, and all seven in RESULT_HANDLE_TOOLS), HANDLE_META_KEY and HANDLE_ID_PATTERN. Check an ID with re.fullmatch(HANDLE_ID_PATTERN, value), because a Python $ also matches before a trailing newline.

Migrating from 0.1.0a1

0.1.0a2 renames the clients so that, following the naming adopted in the ADR-0003 amendment, <Product>Service is the control-plane client and <Product>Client is the data-plane client.

The old ContextGatewayClient name now means the data-plane client, so code that builds it with server_url= fails with a TypeError. Rename those calls to ContextGatewayService. redis_context_gateway.mcp.connect was deprecated in 0.1.0a2 and is removed in the next alpha, so importing it fails with ImportError. The old root modules (models, errors, utils, types, sdk and the rest of the generated code) are gone, so those imports fail with ImportError until they import from redis_context_gateway.controlplane. Operations, models, errors and their arguments are otherwise unchanged.

Safety

• Neither client follows redirects, so the admin token and agent keys stay on the Gateway.
• Plain HTTP is refused unless the host is loopback (localhost, 127.0.0.0/8 or ::1).
• Each request must match its own operation's method, path and query names, so an ID like ../../v1/gateways fails before anything is sent. A custom HTTP client that can't turn redirects off is refused with a TypeError.
• The SDK never reads credentials from environment variables. Pass them in explicitly. The redis-iris CLI and goal objects own IRIS_CONTEXT_GATEWAY_* settings and the [context_gateway] config table.
• Upstream secrets are write-only: no response model carries them.
• Generated payload debugging is disabled, even when debug_logger is supplied, because it would log request secrets. Application logging and custom HTTP transports are outside this guarantee. Do not log whole generated exceptions: validation diagnostics can retain supplied credentials, and response errors retain server-provided bodies.
• No call is retried automatically.

Development

uv venv && uv pip install -e . pytest pytest-asyncio ruff
# Python 3.11+: install the local Lens package for its integration test.
uv pip install ../lens-python-sdk
.venv/bin/python -m pytest tests

Speakeasy generates the control plane straight into src/redis_context_gateway/controlplane/ (moduleName: redis_context_gateway.controlplane in .speakeasy/gen.yaml). Hand-written files, protected by .genignore, are the package root (src/redis_context_gateway/__init__.py and py.typed), the data plane (src/redis_context_gateway/mcp.py), the data model types (src/redis_context_gateway/datamodel.py), the control-plane hooks (src/redis_context_gateway/controlplane/_hooks/registration.py), the connection variant check (src/redis_context_gateway/controlplane/utils/connection_variants.py), tests/ and this README. The patch files in .speakeasy/patches/ add that check to the generated MCPConnection and ContextSurfaceConnection models, and Speakeasy reapplies them on every generation. Everything else is regenerated by speakeasy run -t context-gateway-python-sdk from the repository root, or by the sdk_generation_context_gateway_python_sdk.yaml workflow.