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("you@yourco.com", "...")

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("user@example.com", "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.

---

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.

Namespace	Description
client.answers	Grounded answer synthesis with policy-filtered citations (Team+)
client.api_keys	Create, list, delete, and rotate API keys
client.audit	Audit log listing and CSV export
client.auth	Login, signup, token refresh, logout
client.billing	Plans, usage meters, invoices, subscription, Stripe checkout and portal
client.connectors	Connector CRUD, connection testing, search/ingestion config, coverage, classification suggestions
client.dashboard	Aggregated dashboard statistics with optional sparklines
client.data_catalog	Gated resource listing and metadata updates
client.groups	Read-only groups directory with live member counts
client.identity_providers	Identity provider CRUD and sync (Okta, Azure Entra ID, AWS IAM, GCP)
client.ingest	Single-document, batch, and file ingestion (Tier 1 connectors)
client.onboarding	Onboarding status (6 computed steps) and checklist dismissal
client.pipelines	Pipeline CRUD and run management
client.policies	Policy CRUD, lifecycle (activate/archive), and templates
client.principals	Principal listing, detail, and resolution by email or provider subject
client.relationships	REBAC direct-relation CRUD — create, list, delete 1-hop tuples (Team+)
client.retroactive	Retroactive vector registration for existing connectors
client.retrievals	Permission-gated retrieval execution, policy filter, and history
client.simulator	Dry-run, live-preview, and batch-preview access simulation (Growth+)
client.sources	Content-source connections (Drive, SharePoint, Confluence, Notion): create, test, ACL coverage
client.users	Current user profile — get_me(), update_me(name)

---

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="alice@company.com")

# 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="alice@company.com",
    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:

Endpoint	Limit
POST /api/retrievals/execute	60/min
POST /api/answers/execute	20/min
POST /api/simulator/preview	10/min

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")

---

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

Tool	Description
gateco_retrieve	Permission-aware retrieval (vector/keyword/hybrid/grep)
gateco_ask	Grounded answer synthesis with search modes (Team+)
gateco_check_access	Dry-run access simulation (Growth+)
gateco_list_connectors	List connectors with readiness levels
gateco_list_principals	List identity principals
gateco_resolve_principal	Resolve a principal by email or provider subject

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