Gateco

Gateco Python SDK

Official Python client for the Gateco API — permission-aware retrieval for AI systems.

The problem it solves

Without Gateco, when an employee asks your AI assistant "What is the CEO's salary?", the RAG pipeline returns the salary from a leaked HR document.

With Gateco:

from gateco_sdk import GatecoClient

client = GatecoClient("https://api.gateco.ai")
client.login("[email protected]", "...")

result = client.retrievals.execute(
    query="What is the CEO's salary?",
    principal_id="user_james_wu",
    connector_id="connector_hr_docs",
    search_mode="hybrid",
)

# result.allowed_chunks → [] (denied — James Wu lacks HR classification access)
# result.denied_count   → 1
# result.decision       → "DENIED"
# Your AI model never sees the salary data

Gateco sits between your AI agent and your vector store. Every retrieval is evaluated against your access policies before any content reaches the model.

Installation

pip install gateco

For MCP server support (Claude Desktop, Cursor, etc.):

pip install gateco[mcp]

Authentication

Gateco API keys use the format gck_<env>_<random> (e.g. gck_live_abc123...).

Generate keys via the dashboard or via client.api_keys.create(name="my-service").

from gateco_sdk import AsyncGatecoClient, GatecoClient

# Async client with API key
client = AsyncGatecoClient("https://api.gateco.ai", api_key="gck_live_abc123...")

# Sync client with API key
client = GatecoClient("https://api.gateco.ai", api_key="gck_live_abc123...")

# Or use email/password login (issues a short-lived JWT)
client = GatecoClient("https://api.gateco.ai")
client.login("[email protected]", "password")

The API key is sent as the X-API-Key header on every request. Set it via the GATECO_API_KEY environment variable when using the CLI or MCP server. The key takes precedence over a stored login.

No credential at all? Since 1.13.0 a client constructed with no api_key, no GATECO_API_KEY and no access_token reads the session that gateco login stored in ~/.gateco/credentials.json (and its base_url), and writes refreshed tokens back. One login on a machine serves the CLI, the MCP server and your own scripts; do not copy the file between machines, because refresh rotates and the two copies would revoke each other.

Which routes accept an API key

A key carries explicit scopes, chosen when it is created (Settings > API Keys, every plan). Everything not listed here needs a user session (gateco login or client.login(...)); a key sent there gets 401 AUTH_JWT_REQUIRED, and a key without the needed scope gets 403 API_KEY_SCOPE_MISSING.

Session only: policies, connector create/update/delete and their search and ingestion configs, identity providers, API-key management, audit, retrieval history, billing, the access simulator, classification suggestions.

Quick Start

Async (recommended for production services)

import asyncio
from gateco_sdk import AsyncGatecoClient

async def main():
    async with AsyncGatecoClient(
        "https://api.gateco.ai",
        api_key="gck_live_abc123...",
    ) as client:

        # Policy-gated retrieval — the core Gateco primitive
        result = await client.retrievals.execute(
            query="What is the CEO's salary?",
            principal_id="user_james_wu",
            connector_id="connector_hr_docs",
            search_mode="hybrid",
            alpha=0.7,   # 70% vector weight, 30% keyword
            top_k=5,
        )

        # Allowed chunks are safe to pass to your LLM
        for chunk in result.allowed_chunks:
            print(f"[ALLOWED] {chunk.resource_id} score={chunk.score}")

        # Denied chunks are redacted — only metadata is surfaced
        print(f"Denied: {result.denied_count} chunk(s)")

asyncio.run(main())

Synchronous (scripts and notebooks)

from gateco_sdk import GatecoClient

with GatecoClient("https://api.gateco.ai", api_key="gck_live_abc123...") as client:
    result = client.retrievals.execute(
        query="What is the CEO's salary?",
        principal_id="user_james_wu",
        connector_id="connector_hr_docs",
        search_mode="hybrid",
    )
    print(result.decision)  # "DENIED"

Available Namespaces

Every namespace, and every method on it, is available on both AsyncGatecoClient (async) and GatecoClient (sync); tests/test_sync_parity.py fails the build if the two drift.

Retrieval Search Modes

# Vector search (default) — semantic similarity
result = await client.retrievals.execute(
    query="quarterly earnings", principal_id="...", connector_id="...",
)

# Keyword search — ranked full-text search (BM25)
result = await client.retrievals.execute(
    query="quarterly earnings", principal_id="...", connector_id="...",
    search_mode="keyword",
)

# Hybrid search — vector + keyword fused (RRF)
result = await client.retrievals.execute(
    query="quarterly earnings", principal_id="...", connector_id="...",
    search_mode="hybrid",
    alpha=0.5,   # 1.0 = all-vector, 0.0 = all-keyword
)

# Grep — exact pattern matching
result = await client.retrievals.execute(
    query="ERR-4021", principal_id="...", connector_id="...",
    search_mode="grep",
    pattern_type="regex",
    case_sensitive=False,
)

API Key Management

# Create a key — the plaintext is returned exactly once
key_info = await client.api_keys.create(name="prod-worker")
print(key_info["key"])    # gck_live_abc123...  (store this securely)
print(key_info["prefix"]) # gck_live_abc

# List keys (plaintext never returned after creation)
keys = await client.api_keys.list()

# Rotate a key — old key is invalidated immediately
new_key = await client.api_keys.rotate(key_id="key-uuid-here")

# Delete a key
await client.api_keys.delete(key_id="key-uuid-here")

Relationship-Based Access Control (REBAC)

# Create a direct relation: Alice owns resource R
rel = await client.relationships.create(
    subject_principal_id="principal-uuid",
    relation_name="owner_of",
    object_resource_id="resource-uuid",
)
print(rel["id"])

# List relations for a principal
rels = await client.relationships.list(
    subject_id="principal-uuid",
    relation="owner_of",
)

# Delete a relation (invalidates policy cache immediately)
await client.relationships.delete(relationship_id=rel["id"])

Use relation.<name> as a policy condition field to gate access on the existence of a tuple:

# Policy rule: allow access when principal has owner_of relation on the resource
rule = {"field": "relation.owner_of", "operator": "eq", "value": True}

Onboarding Status

# Check which onboarding steps are complete
status = await client.onboarding.status()
for step in status["steps"]:
    print(f"{step['name']:30s}  {step['status']}")

# Dismiss the checklist once the org is fully configured
await client.onboarding.dismiss()

Principal Resolution

# Resolve a principal by email (read-only — never creates)
principal = await client.principals.resolve(email="[email protected]")

# Resolve by raw IDP-side user ID
principal = await client.principals.resolve(provider_subject="okta-user-123")

# Scoped to a specific identity provider
principal = await client.principals.resolve(
    email="[email protected]",
    identity_provider_id="idp-uuid-here",
)

Grounded Answer Synthesis (Team+)

answer = await client.answers.execute(
    query="Summarise the Q4 revenue results.",
    principal_id="user_alice",
    connector_id="connector_finance_docs",
    search_mode="hybrid",
)

print(answer.answer_text)      # LLM-generated answer from allowed chunks only
print(answer.outcome)          # "answered" | "no_access" | "insufficient_context"
for citation in answer.citations:
    print(f"  [{citation.score:.2f}] {citation.resource_id}")

Policy Creation

# Create an RBAC policy
policy = await client.policies.create(
    name="Engineering read-only",
    description="Allow engineering group to read internal resources",
    type="rbac",
    effect="allow",
    rules=[{
        "description": "Engineering group members",
        "effect": "allow",
        "conditions": [{"field": "principal.groups", "operator": "contains", "value": "engineering"}],
        "priority": 1,
    }],
    resource_selectors=[{"field": "resource.classification", "op": "lte", "value": "internal"}],
)

Policy validation rules:

• Condition fields must use resource., principal., or relation. prefix. Bare field names (e.g., "classification") are rejected with 422 — they silently resolve against the principal rather than the resource.
• Policies with empty resource_selectors require apply_to_all_resources=True in the request body to opt into matching all resources explicitly.

Retrieval Diagnostics

result = await client.retrievals.execute(
    query="quarterly earnings",
    principal_id="user_alice",
    connector_id="connector_finance_docs",
    search_mode="hybrid",
)

# All retrieval responses include diagnostics
print(result.diagnostics.outcome_detail)    # Human-readable explanation
print(result.diagnostics.candidates_fetched)  # How many candidates were checked
print(result.diagnostics.candidates_denied)   # How many were denied by policy
print(result.diagnostics.refill_rounds)       # How many refill rounds ran (0 = first pass sufficient)

Connector Preflight Check

# Check if a connector is production-ready before using it in retrievals
preflight = client.connectors.preflight(connector_id="...")
print(preflight.ready_for_production)  # bool
print(preflight.recommendation)        # What to fix next
for check in preflight.checks:
    print(f"{check.name}: {'PASS' if check.passed else 'FAIL'} (blocking={check.blocking})")

Dashboard Activation Metrics

# Aggregated dashboard statistics
stats = await client.dashboard.stats()
print(stats["total_retrievals"])
print(stats["allowed_retrievals"])

# Activation funnel metrics
activation = client.dashboard.get_activation_stats()
print(activation.total_retrievals_30d)
print(activation.allowed_retrievals_30d)
print(activation.no_access_retrievals_30d)  # Retrievals where 0 results were authorized
print(activation.p95_latency_ms)            # End-to-end p95 latency

Pagination

List endpoints return a Page object. Use list_all() for automatic async pagination:

async for connector in client.connectors.list_all():
    print(connector.name)

Rate Limits

Three endpoints enforce per-org-per-minute limits:

Exceeded limits raise RateLimitError. The SDK retries automatically with exponential backoff (configurable via max_retries). Limits are org-scoped and reset on process restart (in-memory implementation).

Error Handling

from gateco_sdk.errors import NotFoundError, RateLimitError, AuthenticationError

try:
    conn = await client.connectors.get("nonexistent-id")
except NotFoundError:
    print("Connector not found")
except RateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after}s")
except AuthenticationError:
    print("Invalid or expired credentials")

CLI

The gateco command is installed with the SDK and uses the same credentials.

# Log in once; you are prompted for the password (or pipe it on stdin).
# Passing --password on the command line works but lands in shell history.
gateco login --email [email protected]

gateco whoami                       # base URL, credential in use, key scopes

# Connectors: inspect and configure (user session only)
gateco connectors list
gateco connectors get <connector_id>
gateco connectors get-search-config <connector_id>
gateco connectors set-search-config <connector_id> --json '{"table_name": "docs", "vector_column": "embedding", "content_column": "text"}'
gateco connectors get-ingestion-config <connector_id>
gateco connectors set-ingestion-config <connector_id> --file ingestion.json

# Ingest: .txt/.md are sent as text; PDF, Word, PowerPoint, Excel and CSV are uploaded
gateco ingest report.txt --connector-id <id> --classification confidential --sensitivity high
gateco ingest "Q3 board deck.pdf" --connector-id <id> --classification confidential --sensitivity high --domain finance
gateco ingest-batch ./documents --connector-id <id> --glob "*.md"

# Retrieve as a principal (API key with the retrieve scope, or a session)
gateco retrieve --connector-id <id> --principal-id <id> --query "revenue report" --search-mode hybrid

--classification accepts public, internal, confidential, restricted; --sensitivity accepts low, medium, high, critical (gateco ingest --help lists them).

MCP Server (Model Context Protocol)

The optional MCP server lets AI agents (Claude Desktop, Cursor, etc.) perform permission-aware retrieval without any custom code.

pip install gateco[mcp]

# Start the server
gateco mcp serve

# Or use the direct entry point (for MCP host configs)
gateco-mcp

Claude Desktop Configuration

{
  "mcpServers": {
    "gateco": {
      "command": "gateco-mcp",
      "env": {
        "GATECO_API_KEY": "gck_live_abc123...",
        "GATECO_BASE_URL": "https://api.gateco.ai"
      }
    }
  }
}

Available MCP Tools

All tools return markdown-formatted text. Denied content is never exposed — only denial reasons and counts are shown.

Development

pip install -e ".[dev]"
pytest -v

# Run MCP server tests
pytest tests/test_mcp/ -v

# With coverage
pytest --cov=src/gateco_sdk

Links

• Documentation
• Dashboard
• GitHub
• Bug Tracker
• Support