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_ _ (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. 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. 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"