Gateco

Gateco Python SDK

适用于 Gateco API 的官方 Python 客户端 — 为 AI 系统提供基于权限的检索。

---

它解决的问题

如果没有 Gateco，当员工问 AI 助手“CEO 的薪水是多少？”时， RAG 工作流会从泄露的 HR 文档中返回薪水信息。

使用 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 位于 AI Agent 与向量存储之间。每次检索都会先根据你的访问策略进行评估， 再将任何内容发送给模型。

---

安装

pip install gateco

如需支持 MCP 服务（Claude Desktop、Cursor 等）：

pip install gateco[mcp]

---

身份验证

Gateco API 密钥采用 gck_<env>_<random> 格式（例如 gck_live_abc123...）。

可通过控制面板或使用 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")

每次请求都会通过 X-API-Key 标头发送 API 密钥。使用 CLI 或 MCP 服务时，可通过 GATECO_API_KEY 环境变量设置。该密钥的优先级高于已存储的登录凭据。

完全没有凭据？ 从 1.13.0 开始，如果创建客户端时未提供 api_key、未设置 GATECO_API_KEY，也未提供 access_token，客户端会读取 gateco login 存储在 ~/.gateco/credentials.json 中的会话（及其 base_url），并将刷新的令牌写回该文件。在同一台 机器上登录一次即可供 CLI、MCP 服务和你自己的脚本使用；请勿在不同机器间复制该 文件，因为令牌刷新会进行轮换，而两份副本会互相撤销。

哪些路由接受 API 密钥

密钥包含明确的权限范围，在创建时选择（Settings > API Keys，所有套餐均可用）。 此处未列出的所有操作都需要用户会话（gateco login 或 client.login(...)）； 向这些操作发送密钥会得到 401 AUTH_JWT_REQUIRED，而缺少所需权限范围的密钥会得到 403 API_KEY_SCOPE_MISSING。

权限范围	路由
retrieve	POST /api/retrievals/execute、POST /api/retrievals/filter、POST /api/answers/execute、GET /api/principals、GET /api/principals/:id、POST /api/principals/resolve、GET /api/groups（自 2026-10-02 起）、GET /api/connectors（id、name、type、status）
ingest	POST /api/v1/ingest、/batch、/file、/files、DELETE /api/v1/resources/:id、/api/v1/ingest/jobs*
relationships	POST、GET、DELETE /api/relationships
principals	POST、PATCH、DELETE /api/principals（本地目录）
任意密钥	GET /api/users/me（返回密钥名称和权限范围）

仅限会话：策略、连接器的创建／更新／删除及其搜索和摄取配置、身份提供方、API 密钥管理、审计、检索历史、账单、访问模拟器、分类建议。

---

快速开始

异步（推荐用于生产服务）

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

同步（脚本和笔记本）

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"

---

可用命名空间

每个命名空间及其上的每个方法，在 AsyncGatecoClient（异步）和 GatecoClient（同步）上均可用；如果两者出现偏差，tests/test_sync_parity.py 会导致构建失败。

命名空间	描述
client.answers	基于依据的答案综合，提供经过策略过滤的引用（Team+）
client.api_keys	创建、列出、删除和轮换 API 密钥
client.audit	审计日志列表和 CSV 导出
client.auth	登录、注册、令牌刷新、登出
client.billing	套餐、用量计量、发票、订阅、Stripe 结账和门户
client.connectors	连接器的增删改查、连接测试、搜索／摄取配置、覆盖范围、分类建议
client.dashboard	聚合仪表板统计数据，可选提供迷你图
client.data_catalog	受控的资源列表和元数据更新
client.groups	只读群组目录，包含实时成员数
client.identity_providers	身份提供商的增删改查和同步（Okta、Azure Entra ID、AWS IAM、GCP）
client.ingest	单文档、批量和文件摄取（Tier 1 连接器）
client.onboarding	入门状态（6 个计算步骤）和清单关闭
client.pipelines	管道的增删改查和运行管理
client.policies	策略的增删改查、生命周期管理（激活／归档）和模板
client.principals	主体列表、详情，以及按电子邮件或提供商主题标识符解析
client.relationships	REBAC 直接关系的增删改查 — 创建、列出、删除 1 跳元组（Team+）
client.retroactive	为现有连接器进行追溯向量注册
client.retrievals	受权限控制的检索执行、策略过滤和历史记录
client.simulator	访问模拟的试运行、实时预览和批量预览（Growth+）
client.sources	内容源连接（Drive、SharePoint、Confluence、Notion）：创建、测试、ACL 覆盖范围
client.users	当前用户个人资料 — get_me()、update_me(name)

---

检索搜索模式

# 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 密钥管理

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

---

基于关系的访问控制（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"])

将 relation.<name> 用作策略条件字段，以便根据元组是否存在来控制访问：

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

---

入门状态

# 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()

---

主体解析

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

---

基于依据的答案综合（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}")

---

策略创建

# 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"}],
)

策略验证规则：

• 条件字段必须使用 resource.、principal. 或 relation. 前缀。 不带前缀的字段名（例如 "classification"）会被拒绝并返回 422 — 它们会静默地解析为主体字段，而不是资源字段。
• resource_selectors 为空的策略，必须在请求正文中设置 apply_to_all_resources=True，以明确选择匹配所有资源。

---

检索诊断

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)

---

连接器预检

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

---

仪表板激活指标

# 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

---

分页

列表端点会返回一个 Page 对象。使用 list_all() 可自动进行异步分页：

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

---

速率限制

三个端点实施每个组织每分钟的请求限制：

端点	限制
POST /api/retrievals/execute	60/min
POST /api/answers/execute	20/min
POST /api/simulator/preview	10/min

超出限制会引发 RateLimitError。SDK 会使用指数退避自动重试（可通过 max_retries 配置）。限制按组织设定，并在进程重启时重置（内存实现）。

---

错误处理

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

---

命令行界面

SDK 会安装 gateco 命令，并使用相同的凭据。

# 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 接受 public、internal、confidential、restricted； --sensitivity 接受 low、medium、high、critical（gateco ingest --help 会列出这些选项）。

---

MCP 服务（Model Context Protocol）

可选的 MCP 服务可让 AI Agent（Claude Desktop、Cursor 等）无需任何自定义代码， 即可执行权限感知检索。

pip install gateco[mcp]

# Start the server
gateco mcp serve

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

Claude Desktop 配置

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

可用的 MCP 工具

工具	描述
gateco_retrieve	权限感知检索（向量／关键词／混合／grep）
gateco_ask	基于搜索模式生成有依据的回答（Team+）
gateco_check_access	访问权限模拟（试运行）（Growth+）。仅限用户会话（gateco login）；不能使用 API 密钥调用
gateco_list_connectors	列出连接器及其就绪级别
gateco_list_principals	列出身份主体
gateco_list_groups	列出群组（IdP 同步及本地目录），并显示实时成员数。使用 retrieve 范围密钥连接 2026-10-02 起的后端时可用；较旧的后端会提示需要用户会话
gateco_resolve_principal	根据电子邮件或提供方主题解析主体

所有工具均返回 Markdown 格式的文本。绝不会暴露被拒绝访问的内容——只会显示拒绝原因和数量。

---

开发

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

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

# With coverage
pytest --cov=src/gateco_sdk

---

链接

• 文档
• 控制面板
• GitHub
• 问题跟踪器
• 支持