redis-context-retriever 0.0.5

Python client for the Context Surfaces API — manage context surfaces and invoke MCP tools. Includes an optional CLI (ctxctl).

Redis Context Retriever Python Client

redis-context-retriever is the Python client and ctxctl CLI for the Redis Context Retriever API. The API's core resource is a context surface — a configurable layer that connects your data in Redis to MCP tools. This package can both manage context surfaces and speak MCP (Model Context Protocol) directly so agents can invoke the tools surfaces generate.

Contents

• What is a context surface
• When to use this package
• Key features
• Installation
• Common concepts
• Python client
• CLI
• Identity Provider (OAuth) Configuration
• Verify setup
• License

What is a context surface

A context surface is a configurable layer that connects your data (often in Redis) to MCP tools so applications and agents can query, filter, and retrieve information through a consistent API. You use context surfaces to define data models, generate tools, and securely operate those tools with agent keys.

When to use this package

Use redis-context-retriever when you need to:

• Provision and manage context surfaces and admin/agent keys
• Define data models that generate MCP tools
• Invoke MCP tools from Python or the CLI
• Configure OAuth identity providers for user-level access control

Key features

• Complete API coverage for admin keys, context surfaces, agent keys, and MCP tools
• OAuth identity provider integration for user-level access control
• Async-first client built on httpx with retries
• Type-safe models via Pydantic v2
• CLI (ctxctl) with secure credential storage and rich output formats
• Unified client for API + MCP workflows

Installation

Requires Python 3.11+.

pip install redis-context-retriever

Optional with uv:

uv pip install redis-context-retriever

Common concepts

• Provisioning auth: Use an admin API key to create and manage resources.
• Agent auth: Agent keys are used to invoke MCP tools.

Python client

Quickstart

import asyncio
from context_surfaces import ContextSurfacesClient

async def main():
    async with ContextSurfacesClient() as client:
        health = await client.health()
        print(health.status)

asyncio.run(main())

Quickstart (Unified API + MCP)

import os

from context_surfaces import UnifiedClient

async with UnifiedClient() as client:
    admin_key = os.getenv("CTX_ADMIN_KEY")
    if not admin_key:
        raise RuntimeError("Set CTX_ADMIN_KEY before running this example.")
    surface = await client.create_context_surface(admin_key, "My Surface", data_model={...})
    agent_key = await client.create_agent_key(admin_key, surface.id, "Agent")
    result = await client.query_tool(agent_key.key, "search_customer_by_text", {"query": "john"})

Data model example (models.py)

Context surfaces can generate tools from a data model. A minimal models.py might look like:

from context_surfaces.context_model import ContextField, ContextModel


class Customer(ContextModel):
    __redis_key_template__ = "customer:{id}"

    id: int = ContextField(description="Customer ID", is_key_component=True)
    email: str = ContextField(description="Email address", index="text")
    city: str = ContextField(description="City", index="tag")

You can pass this file to the CLI with --models ./models.py, or use it in Python to generate a data_model for create_context_surface.

Import data with UnifiedClient (recommended)

Use UnifiedClient.import_data(...) to ingest validated ContextModel records through the API. This keeps key templates, schema validation, and indexes aligned with your context surface configuration. Avoid writing records directly to Redis unless you are intentionally managing the full storage/index contract yourself.

import asyncio
import os

from context_surfaces import UnifiedClient
from models import Customer


async def main() -> None:
    admin_key = os.getenv("CTX_ADMIN_KEY")
    surface_id = os.getenv("CTX_SURFACE_ID")
    if not admin_key or not surface_id:
        raise RuntimeError("Set CTX_ADMIN_KEY and CTX_SURFACE_ID before running this example.")

    records = [
        Customer(id=1, email="[e-mail]", city="Berlin"),
        Customer(id=2, email="[e-mail]", city="Madrid"),
    ]

    async with UnifiedClient() as client:
        result = await client.import_data(
            admin_key=admin_key,
            context_surface_id=surface_id,
            records=records,
            on_conflict="overwrite",
            on_error="fail_fast",
        )
        print(f"Imported: {result.imported}, Failed: {result.failed}")


asyncio.run(main())

Create a context surface with a Redis data source (Python)

import os

from context_surfaces import (
    ContextSurfacesClient,
    CreateContextSurfaceRequest,
    DataSourceRequest,
    DataSourceConnectionConfig,
)

async with ContextSurfacesClient() as client:
    admin_key = os.getenv("CTX_ADMIN_KEY")
    if not admin_key:
        raise RuntimeError("Set CTX_ADMIN_KEY before running this example.")

    surface = await client.create_context_surface(
        CreateContextSurfaceRequest(
            name="Support",
            data_model={...},
            data_source=DataSourceRequest(
                connection_config=DataSourceConnectionConfig(
                    addr="redis-12345.example.redis.cloud:6380",
                    username="default",
                    password=os.getenv("REDIS_PASSWORD", ""),
                    db=0,
                    tls_enabled=True,
                ),
            ),
        ),
        admin_key=admin_key,
    )

Authentication (Python clients)

For both ContextSurfacesClient and UnifiedClient, authentication is API-key based. The admin key is passed as an argument to each admin operation and is sent as the X-API-Key request header:

import os

from context_surfaces import ContextSurfacesClient, CreateContextSurfaceRequest

async with ContextSurfacesClient() as client:
    admin_key = os.getenv("CTX_ADMIN_KEY")
    if not admin_key:
        raise RuntimeError("Set CTX_ADMIN_KEY before running this example.")
    surface = await client.create_context_surface(
        CreateContextSurfaceRequest(name="Support", data_model={...}),
        admin_key=admin_key,
    )

Use an existing admin key created outside the Python client, for example via Redis Cloud console or ctxctl.

CLI

Quickstart

# Verify installation
ctxctl --help

# Authenticate with an admin key (stored in OS keyring)
ctxctl auth login --admin-key <key>

# Create a context surface with a Redis data source
ctxctl surface create --name "Support" --models ./models.py \
  --redis-addr "redis.example.com:6379" --redis-password "$REDIS_PASSWORD"

# Create an agent key for the surface (save the key from output!)
ctxctl agent create --name "Support Agent" --surface-id <surface-id>

# List MCP tools and call one
ctxctl tools list --agent-key <agent-key>
ctxctl tools call search_customer_by_text --query john --limit 10

# Filter query example (use exact parameter names from the tool schema)
ctxctl tools call filter_customer_by_city \
  --agent-key <key> \
  --value Berlin \
  --limit 5

When calling tools, use argument names exactly as defined by the tool input schema. If a schema field uses underscores, pass the same underscore form in CLI flags.

Create a context surface with a Redis data source (CLI)

# Create surface with a Redis data source
ctxctl surface create --name "Support" --models ./models.py \
  --redis-addr "redis-12345.example.redis.cloud:6380" \
  --redis-username "default" \
  --redis-password "$REDIS_PASSWORD" \
  --redis-tls

# Create an agent key for the surface
ctxctl agent create \
  --name "Support Agent" \
  --surface-id <surface-id>

Important: Save the agent key output securely - it cannot be retrieved later!

You can also pass the admin key explicitly instead of using stored credentials:

ctxctl agent create \
  --name "Support Agent" \
  --surface-id <surface-id> \
  --admin-key <admin-key>

Optional parameters for agent keys:

# With description
ctxctl agent create \
  --name "Analytics Agent" \
  --surface-id <surface-id> \
  --description "Agent for analytics dashboard"

# With access tags for multi-tenant filtering
ctxctl agent create \
  --name "Tenant Agent" \
  --surface-id <surface-id> \
  --access-tags '{"tenant": ["acme"]}'

Authentication and configuration

• Admin keys or Redis Cloud sessions (Service Manager credentials) are required for provisioning resources.
• Agent keys are used for querying MCP tools.
• CLI credentials are stored securely in the OS keyring and can be overridden via env vars.

Common configuration options:

# Config file location
export CTX_CONFIG_FILE=~/.config/ctxctl/config.yaml

# Auth override
export CTX_ADMIN_KEY=<key>

You can inspect and update config with ctxctl config show and ctxctl config set <key> <value>.

Redis Cloud session login (CLI)

If you have Redis Cloud credentials, you can create a session and use it for admin operations:

ctxctl auth login --username [e-mail]
ctxctl admin create --name "My Admin Key"

Identity Provider (OAuth) Configuration

Context surfaces support OAuth-based user access control. Configure an external identity provider (IdP) to validate JWT tokens and extract user-specific access tags from claims.

Configure IdP via CLI

# Basic IdP setup (auto-discovers JWKS from issuer)
ctxctl surface idp configure <surface-id> \
  --issuer "https://auth.example.com/realms/my-realm"

# With audience and access tag claim mappings
ctxctl surface idp configure <surface-id> \
  --issuer "https://auth.example.com/realms/my-realm" \
  --audience "my-client-id" \
  --claim-mapping "department=department" \
  --claim-mapping "tenant=tenant_id"

# Using provider shortcut (keycloak, auth0)
ctxctl surface idp configure <surface-id> \
  --provider keycloak \
  --issuer "https://keycloak.example.com/realms/my-realm"

# Import from JSON file
ctxctl surface idp configure <surface-id> --from-file ./idp-config.json

Other IdP commands

# Show current IdP configuration
ctxctl surface idp show <surface-id>

# Test IdP connectivity and optionally validate a token
ctxctl surface idp test <surface-id>
ctxctl surface idp test <surface-id> --token "eyJhbGciOi..."

# Disable/Enable IdP (keeps config but toggles validation)
ctxctl surface idp disable <surface-id>
ctxctl surface idp enable <surface-id>

Configure IdP via Python

from context_surfaces import (
    ContextSurfacesClient,
    IdentityProviderConfig,
    UpdateContextSurfaceRequest,
)

async with ContextSurfacesClient() as client:
    idp_config = IdentityProviderConfig(
        enabled=True,
        issuer="https://auth.example.com/realms/my-realm",
        jwks_uri="https://auth.example.com/realms/my-realm/protocol/openid-connect/certs",
        audience="my-client-id",
        user_claim="sub",
        access_tag_claims={"tenant": "tenant_id"},
        require_user_token=True,
    )

    await client.update_context_surface(
        surface_id,
        UpdateContextSurfaceRequest(identity_provider=idp_config),
        admin_key=admin_key,
    )

Security Note

The allow_insecure option (for HTTP URLs instead of HTTPS) is only available in development environments. In production, all IdP URLs must use HTTPS.

Verify setup

After provisioning and import, validate the full path in a few quick checks:

1. Confirm tools are available:

ctxctl tools list --agent-key <agent-key>

2. Run one filter query:

ctxctl tools call filter_customer_by_city \
  --agent-key <agent-key> \
  --value Berlin \
  --limit 5

3. Validate the response shape includes result metadata and records: count, results, and pagination fields such as has_more/total_count.

Common workflows

1. Create an admin key via Redis Cloud console or ctxctl
2. Create a context surface with a data source
3. Create an agent key for the surface
4. Invoke MCP tools with the agent key

License

MIT License