Gateco Sdk Python MCP Server

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(api_key="gck_live_abc123...")

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

All 19 namespaces are available on both AsyncGatecoClient (async) and GatecoClient (sync).

---

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:

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

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