Gateco
Permission-aware retrieval for AI systems: policy-enforced access to organizational knowledge.
Open source Open in the app JSON README (API)
About
Permission-aware retrieval for AI systems: policy-enforced access to organizational knowledge.
Details
- Kind
- MCP servers
- Topic
- No topic detected
- Publisher
- ai.gateco
- Origin
- official
- Category
- ferramentas
- Transport
- local
- Version
- 1.8.1
- Last push
- 2026-09-04T14:02:35Z
- Repository state
- ativo
- Language
- Python
- License
- MIT
- Added
- 2026-08-29 03:00:11
- Updated
- 2026-08-29 03:00:11
- Origin id
ai.gateco/gateco
README
# Gateco Python SDK
Official Python client for the [Gateco](https://gateco.ai) API — permission-aware retrieval for AI systems.
[](https://pypi.org/project/gateco/)
[](https://www.python.org/)
[](https://github.com/fortisil/gateco-sdk-python)
<!-- mcp-name: ai.gateco/gateco -->
---
## 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:
```python
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
```bash
pip install gateco
```
For MCP server support (Claude Desktop, Cursor, etc.):
```bash
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")`.
```python
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)
```python
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)
```python
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
```python
# 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
```python
# 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)
```python
# 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:
```python
# Policy rule: allow access when principal has owner_of relation on the resource
rule = {"field": "relation.owner_of", "operator": "eq", "value": True}
```
---
## Onboarding Status
```python
# 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
```python
# 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+)
```python
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
```python
# 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
```python
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
```python
# 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
```python
# 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:
```python
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
```python
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.
```bash
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
```json
{
"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
```bash
pip install -e ".[dev]"
pytest -v
# Run MCP server tests
pytest tests/test_mcp/ -v
# With coverage
pytest --cov=src/gateco_sdk
```
---
## Links
- [Documentation](https://gateco.ai/docs)
- [Dashboard](https://app.gateco.ai)
- [GitHub](https://github.com/fortisil/gateco-sdk-python)
- [Bug Tracker](https://github.com/fortisil/gateco-sdk-python/issues)
- [Support](mailto:support@gateco.ai)