IdentArk Gateway

identark

AgentGateway 协议——安全、可扩展的 AI Agent 执行基础设施。

---

问题

AI Agent 能够执行代码、调用 API 或访问文件时，就会在进程中运行。该进程拥有一个运行环境，其中通常包含可能造成严重损害的一切信息：大模型 API 密钥、数据库凭据、AWS 令牌。

最简单的解决方案——让 Agent 与 REST API 在同一个后端运行——会同时引发两个问题：

1. 安全性：Agent 可以访问机器上的所有机密信息。
2. 可靠性：耗费大量内存的 Agent 会拖慢 API。重新部署 API 会终止所有正在运行的 Agent。

identark 可以同时解决这两个问题。

---

工作原理

该 SDK 实现了 AgentGateway 协议——在 Agent 逻辑与外部世界之间提供简洁的接口。开箱即用地提供两种实现：

网关	适用场景	凭据	历史记录
DirectGateway	本地开发、CI 评估	你的 API 密钥	内存中
ControlPlaneGateway	在 IdentArk 上用于生产环境	零——Agent 中没有凭据	控制平面数据库

在这两种环境中，你的 Agent 代码完全相同。只需更改两行代码即可切换。

---

快速开始

pip install identark[openai]

import asyncio
from openai import AsyncOpenAI
from identark import DirectGateway, Message, Role

async def main():
    gateway = DirectGateway(
        llm_client=AsyncOpenAI(),   # Your API key — not in the agent loop
        model="gpt-4o",
    )

    response = await gateway.invoke_llm(
        new_messages=[Message(role=Role.USER, content="Hello, IdentArk!")]
    )

    print(response.message.content)
    print(f"Cost: ${response.cost_usd:.6f}")

asyncio.run(main())

切换到生产环境

更改两行代码即可。无需改动 Agent 逻辑。

# Before (local)
from identark import DirectGateway
gateway = DirectGateway(llm_client=AsyncOpenAI(), model="gpt-4o")

# After (production — agent holds zero secrets)
from identark import ControlPlaneGateway
gateway = ControlPlaneGateway()  # auto-detects env vars inside a IdentArk sandbox

受管控的操作（付款、付款链接、退款）

Agent 请求控制平面执行操作；服务提供商密钥（Stripe、Paystack、 GitHub 或 AWS）保留在 IdentArk 中。执行任何操作前都会应用授权、预算和人工审批， 且重复使用的幂等键不会导致操作执行两次。

request = await gateway.request_capability(
    "payments.payment_link.create",
    "paystack",
    {"amount_minor": 1_500_000, "currency": "NGN",
     "customer_email": "[email protected]", "reference": "order-1001"},
    idempotency_key="order-1001-link",
)
# "succeeded" returns the checkout link; "pending_approval" waits for a person.
execution = await gateway.get_capability_execution(request["request_id"])

---

安装

# Core SDK only
pip install identark

# With OpenAI support
pip install identark[openai]

# With Anthropic support
pip install identark[anthropic]

# With Google Gemini support
pip install identark[gemini]

# With Mistral AI support (EU provider)
pip install identark[mistral]

# All cloud providers
pip install identark[all]

要求： Python 3.10+

使用 TypeScript？同等功能的 SDK 以零运行时依赖的形式提供， 即 identark npm package，并具备相同的 AgentGateway 契约和结构化凭据会话。

---

数据主权

IdentArk 从设计之初就支持任何大模型提供商，包括那些 会将数据保留在英国或欧盟境内的提供商。AgentGateway 协议将 Agent 逻辑与 推理提供商解耦——切换提供商只需更改一行代码。

使用 Ollama 完全本地运行（数据零外传）

from openai import AsyncOpenAI
from identark import DirectGateway

gateway = DirectGateway(
    llm_client=AsyncOpenAI(
        base_url="http://localhost:11434/v1",
        api_key="ollama",
    ),
    model="llama3.2",
    provider="local",   # forces $0 cost tracking; inference stays on your machine
)

安装 Ollama：brew install ollama && ollama pull llama3.2 && ollama serve

使用 Mistral AI（欧盟数据驻留）

from openai import AsyncOpenAI
from identark import DirectGateway

gateway = DirectGateway(
    llm_client=AsyncOpenAI(
        base_url="https://api.mistral.ai/v1",
        api_key="your-mistral-api-key",
    ),
    model="mistral-small-latest",   # auto-detected as "mistral" provider
)

Mistral AI 是一家法国公司。所有推理都在欧盟数据中心运行，并受欧盟数据 保护法（GDPR）约束。如果英国或欧盟的数据治理要求禁止将推理流量发送到 美国云服务提供商，请使用此方案。

完整的可运行脚本请参阅 examples/。

---

AgentGateway 协议

任何实现以下四个异步方法的类都是有效的网关：

class AgentGateway(Protocol):
    async def invoke_llm(self, new_messages, tools=None, tool_choice="auto") -> LLMResponse: ...
    async def persist_messages(self, messages) -> None: ...
    async def request_file_url(self, file_path, method="PUT") -> PresignedURL: ...
    async def get_session_cost(self) -> float: ...

让 Agent 基于该协议编写。具体实现——本地或生产环境——属于运行时细节。

---

功能

• 无需密钥的 Agent — ControlPlaneGateway 不持有 API 密钥、数据库凭据或云令牌
• 为无状态而设计 — 对话历史由网关持有，而非 Agent；终止并重启也不会丢失数据
• 与框架无关 — 适用于 LangChain、LlamaIndex、原始 API 调用或任何自定义 Agent 框架
• 内置成本跟踪 — 每次 invoke_llm 调用都会返回 cost_usd；get_session_cost() 会返回累计总额
• 支持 OpenAI + Anthropic — DirectGateway 开箱即支持这两家服务商
• 用于测试的 MockGateway — 测试套件中无需调用大模型；可完整记录调用以供断言
• 完整类型注解 — 带有 py.typed 标记；支持 mypy 严格模式

---

测试你的 Agent

from identark.testing import MockGateway
from identark.models import LLMResponse, Message, Role

async def test_my_agent():
    mock = MockGateway()
    mock.queue_response(LLMResponse(
        message=Message(role=Role.ASSISTANT, content="The answer is 42."),
        cost_usd=0.001,
        model="mock",
        finish_reason="stop",
    ))

    result = await my_agent(gateway=mock)

    assert mock.invoke_llm_call_count == 1
    assert mock.total_messages_sent == 1

---

支持的服务商

服务商	数据驻留地	DirectGateway	GeminiGateway	ControlPlaneGateway
OpenAI（gpt-4o、gpt-4o-mini 等）	美国	✓	—	✓
Anthropic（Claude 模型）	美国	✓	—	✓
Google Gemini	不定	✓*	✓	规划中
Mistral AI	不定	✓	—	✓
Kimi / Moonshot	不定	✓*	—	✓
Azure OpenAI	配置的 Azure 区域	✓*	—	✓
AWS Bedrock	配置的 AWS 区域	—	—	✓
OpenRouter	取决于服务商	✓*	—	✓
Ollama	本地 🏠	✓	—	非托管路由
任何兼容 OpenAI 的端点	不定	✓	—	✓（自定义端点）

*通过兼容 OpenAI 的客户端／基础 URL 使用。要使用 Gemini 原生 SDK 功能，请使用 GeminiGateway。

---

错误处理

from identark.exceptions import CostCapExceededError, RateLimitError, IdentArkError

try:
    response = await gateway.invoke_llm(new_messages=[...])
except CostCapExceededError as e:
    print(f"Cost cap of ${e.cap_usd} reached. Spent: ${e.consumed_usd}")
except RateLimitError as e:
    await asyncio.sleep(e.retry_after_seconds)
except IdentArkError as e:
    # Catch-all for any SDK error
    raise

完整异常层级：IdentArkError > GatewayError > ControlPlaneError > AuthenticationError | CostCapExceededError | SessionNotFoundError

---

架构

┌─────────────────────────────────────┐
│            Your Agent Code          │
│   (depends only on AgentGateway)    │
└──────────────┬──────────────────────┘
               │
    ┌──────────▼──────────┐
    │    AgentGateway      │  ← Protocol (interface)
    │      Protocol        │
    └──────┬────────┬──────┘
           │        │
  ┌────────▼─┐  ┌───▼──────────────┐
  │  Direct  │  │  ControlPlane    │
  │ Gateway  │  │    Gateway       │
  │          │  │                  │
  │ Local /  │  │   Production     │
  │  Evals   │  │  (zero secrets)  │
  └──────────┘  └────────┬─────────┘
                         │ HTTP
                ┌────────▼─────────┐
                │  IdentArk        │
                │  Control Plane   │
                │  (holds creds)   │
                └──────────────────┘

---

社区

• 讨论：GitHub 讨论区 — 提问、分享想法
• 问题：GitHub Issues — 报告错误和提出功能请求
• 在线演示：identark.io/demo — 在浏览器中试用 IdentArk

---

参与贡献

欢迎贡献。提交重大更改前，请先创建 Issue。

git clone https://github.com/identark/identark.git
cd identark
pip install -e ".[dev]"
pre-commit install
pytest tests/unit/

完整指南请参阅 CONTRIBUTING.md。

---

路线图

• LangChain 适配器（IdentArkChatModel）
• LlamaIndex 适配器（IdentArkLLM）
• 流式支持（invoke_llm_stream）
• CrewAI 集成
• LangGraph 集成（IdentArkNode、IdentArkStreamNode）
• 可插拔推理后端（分布式计算）
• 用于一键部署控制平面的 identark-cli

---

许可证

IdentArk SDK 采用 MIT 许可证 — 可免费用于任何用途，包括商业项目和闭源项目。请参阅 LICENSE。

IdentArk 控制平面（托管服务）为专有软件。SDK 可与任何 AgentGateway 后端配合使用，包括完全可自托管的后端。

---

基于 How We Built Secure, Scalable Agent Sandbox Infrastructure 中介绍的控制平面模式构建。