Install
The core package has no runtime dependencies. Provider clients are optional extras.
pip install "identark[openai]" # or [anthropic], [mistral], [gemini], …
The AgentGateway Protocol
Any object implementing these methods is a valid gateway. Code your agent against the protocol; swap the implementation to change environment.
| Method | Purpose |
|---|---|
invoke_llm(new_messages, …) |
One governed completion. Returns message, token usage, and cost_usd. |
invoke_llm_stream(…) |
Streaming variant; yields StreamChunks. |
persist_messages(messages) |
Append to durable session history (control plane DB in prod). |
request_file_url(…) |
Get a pre-signed URL for file I/O without holding cloud creds. |
get_session_cost() |
Authoritative running spend for the session. |
DirectGateway — local development
Your provider key stays out of the agent loop, and you get cost accounting for free.
import asyncio
from openai import AsyncOpenAI
from identark import DirectGateway, Message, Role
async def main():
gateway = DirectGateway(llm_client=AsyncOpenAI(), model="gpt-4o")
resp = await gateway.invoke_llm(
new_messages=[Message(role=Role.USER, content="Summarise ticket #42")]
)
print(resp.message.content)
print(f"cost: ${resp.cost_usd:.6f}")
asyncio.run(main())
ControlPlaneGateway — production
Zero secrets in the agent. The gateway auto-detects its configuration from the environment.
from identark import ControlPlaneGateway, Message, Role
gateway = ControlPlaneGateway() # reads env, see below
resp = await gateway.invoke_llm(
new_messages=[Message(role=Role.USER, content="Summarise ticket #42")]
)
print(resp.message.content, resp.cost_usd)
You can also pass these to the constructor directly:
gateway = ControlPlaneGateway(
api_key="csk_…",
url="https://api.identark.io",
session_id="sess_…",
)
Framework integrations
Drop-in adapters make the gateway the model / execution layer, so governance is transparent to your orchestration code.