SDKs & API
Python SDK
Guard, mask and log LLM calls from Python with agentid-sdk, including OpenAI and LangChain integrations and an async client.
agentid-sdk for Python (version 0.1.27) supports Python 3.9 or later.
Install
pip install agentid-sdkFor OpenAI or LangChain:
pip install agentid-sdk openai
pip install agentid-sdk openai langchain langchain-openaiexport AGENTID_API_KEY="sk_live_..."
export AGENTID_SYSTEM_ID="00000000-0000-0000-0000-000000000000"Initialize
from agentid import AgentID
agent = AgentID() # reads AGENTID_API_KEYapi_keystrAPI key. Defaults to AGENTID_API_KEY.
base_urlstrDefaults to https://app.getagentid.com/api/v1.
strict_modeboolFail-closed on Guard connectivity failures. Default False.
failure_mode"fail_open" | "fail_close"Override the system's failure mode.
pii_maskingboolForce local masking on; omit to follow the dashboard.
client_fast_failboolLocal preflight before Guard. Default False.
guard_timeout_sfloatGuard timeout, default 30.0.
ingest_timeout_sfloatIngest timeout, default 5.0.
AsyncAgentID has the same options and async methods for asyncio applications.
Wrap the OpenAI client
import os
from openai import OpenAI
from agentid import AgentID, SecurityBlockError
agent = AgentID()
client = agent.wrap_openai(
OpenAI(),
system_id=os.environ["AGENTID_SYSTEM_ID"],
user_id="customer-123",
)
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "What is the capital of the Czech Republic?"}],
)
print(response.choices[0].message.content)
except SecurityBlockError as exc:
print("Blocked by AgentID:", exc.reason)wrap_openai(client, system_id, user_id=None, api_key=None, expected_languages=None, request_identity=None, telemetry=None, deanonymize_output_for_client=None) needs openai 1.x or later. Pass per-call telemetry with the extra agentid_telemetry= argument of create; it is removed before the request reaches OpenAI.
The wrapper's pre-flight guard() input is the last user message.
Manual guard and log
import os
from agentid import AgentID
agent = AgentID()
system_id = os.environ["AGENTID_SYSTEM_ID"]
verdict = agent.guard(
input=user_message,
system_id=system_id,
model="gpt-4o-mini",
user_id="customer-123",
)
if not verdict.get("allowed", False):
raise RuntimeError(f"Blocked: {verdict.get('reason')}")
protected_input = verdict.get("transformed_input") or user_message
answer = call_model(protected_input)
agent.log(
system_id=system_id,
event_id=verdict.get("client_event_id"),
model="gpt-4o-mini",
input=protected_input,
output=answer.text,
usage=answer.usage,
)guard() accepts input, system_id, model, user_id, client_event_id, expected_languages, request_identity, metadata, api_key and client_capabilities. The Python SDK does not send attachments or prompt_context; use the Guard API directly for those.
Masking and restoration
- The wrappers follow the dashboard's masking switches.
pii_masking=Trueforces local masking on. - With Restore masked values for the end user on, the wrapper restores placeholders in the response returned to your code. Opt out with
deanonymize_output_for_client=False. - For custom routes, protect the exact history you send:
from agentid import protect_message_history, protect_chat_state
protected = protect_message_history(body["messages"])
response = openai.chat.completions.create(model="gpt-4o-mini", messages=protected.messages)
chat = protect_chat_state(
body["messages"],
transcript={"ui": "raw", "provider": "masked", "persistence": "masked"},
)PIIManager().anonymize(text) returns (masked_text, mapping); deanonymize(text, mapping) restores it.
Optional extras add heavier local detectors: pip install "agentid-sdk[pii]" (then python -m spacy download en_core_web_lg) and pip install "agentid-sdk[security]".
LangChain
import os
from langchain_openai import ChatOpenAI
from agentid import AgentID, AgentIDCallbackHandler
agent = AgentID()
handler = AgentIDCallbackHandler(agent, system_id=os.environ["AGENTID_SYSTEM_ID"])
llm = ChatOpenAI(model="gpt-4o-mini")
print(llm.invoke("Hello", config={"callbacks": [handler]}).content)Agent workflows
from agentid import (
AgentID,
create_agentid_correlation_id,
create_agentid_telemetry_context,
create_agentid_workflow_trail,
)
agent = AgentID()
trail = create_agentid_workflow_trail(
agent=agent,
system_id=os.environ["AGENTID_SYSTEM_ID"],
telemetry=create_agentid_telemetry_context(
{"workflowRunId": create_agentid_correlation_id(), "workflowName": "Candidate intake"}
),
)
trail.run_step(
{"telemetry": create_agentid_telemetry_context({"workflowStepName": "screen_candidate", "eventCategory": "tool"})},
lambda: screen_candidate(),
)For a single event use agent.log_operation(system_id, telemetry=..., event_category="tool", event_status="completed").
Errors
| Error | When |
|---|---|
SecurityBlockError (reason) | A wrapped call or LangChain run was blocked |
SecurityViolationException (reason) | Local enforcement blocked or redacted by policy |
DependencyError (reason, status_code, dependency) | A dependency such as ingest failed in fail-closed mode |