agentid-sdk

agentid-sdk (Python)

1. Introduction

agentid-sdk is the official Python SDK for AgentID, an AI security and compliance System of Record. It lets you enforce guardrails before model execution, capture immutable telemetry for auditability, and integrate security checks into OpenAI and LangChain workflows with minimal code.

The Mental Model

AgentID sits between your application and the LLM runtime:

User Input -> guard() -> [AgentID Policy] -> verdict
                              | allowed
                              v
                         LLM Provider
                              v
                           log() -> [Immutable Ledger]

• guard(): evaluates prompt and context before model execution.
• Model call: executes only if guard verdict is allowed.
• log(): persists immutable telemetry (prompt, output, latency) for audit and compliance.

2. Installation

pip install agentid-sdk

Optional extras:

pip install "agentid-sdk[pii]"
pip install "agentid-sdk[security]"

If you enable Presidio/spaCy-backed PII detection, install the spaCy language model:

pip install "agentid-sdk[pii]"
python -m spacy download en_core_web_lg

3. Prerequisites

1. Create an AgentID account at https://app.getagentid.com.
2. Create an AI system and copy:
   • AGENTID_API_KEY (for example sk_live_...)
   • AGENTID_SYSTEM_ID (UUID)
3. If using OpenAI/LangChain, set:
   • OPENAI_API_KEY

export AGENTID_API_KEY="sk_live_..."
export AGENTID_SYSTEM_ID="00000000-0000-0000-0000-000000000000"
export OPENAI_API_KEY="sk-proj-..."

Compatibility

• Node.js: v18+ / Python: 3.9+ (cross-SDK matrix)
• Thread Safety: AgentID clients are thread-safe and intended to be instantiated once and reused across concurrent requests.
• Latency: async log() is non-blocking for model execution paths; sync guard() typically adds network latency (commonly ~50-100ms, environment-dependent).

4. Quickstart

import os
from agentid import AgentID

agent = AgentID()  # auto-loads AGENTID_API_KEY
system_id = os.environ["AGENTID_SYSTEM_ID"]

verdict = agent.guard(
    input="Summarize this support ticket.",
    system_id=system_id,
    model="gpt-4o-mini",
    user_id="quickstart-user",
)
if not verdict.get("allowed", False):
    raise RuntimeError(f"Blocked: {verdict.get('reason')}")

agent.log(
    system_id=system_id,
    input="Summarize this support ticket.",
    output="Summary generated.",
    model="gpt-4o-mini",
    event_id=verdict.get("client_event_id"),
    metadata={"agent_role": "support-assistant"},
)
print("Guard allowed + telemetry logged")

5. Core Integrations

OpenAI Wrapper

import os
from openai import OpenAI
from agentid import (
    AgentID,
    SecurityBlockError,
    create_agentid_correlation_id,
    create_agentid_telemetry_context,
)

agent = AgentID()
workflow_run_id = create_agentid_correlation_id()
openai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
secured = agent.wrap_openai(
    openai,
    system_id=os.environ["AGENTID_SYSTEM_ID"],
    user_id="customer-123",
    telemetry=create_agentid_telemetry_context(
        {
            "workflowRunId": workflow_run_id,
            "workflowStepName": "answer_question",
            "toolName": "openai.chat",
            "eventCategory": "ai",
        }
    ),
)

try:
    response = secured.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "What is the capital of the Czech Republic?"}],
        agentid_telemetry={
            "workflowStepName": "answer_question_openai",
            "eventSubtype": "answer_generated",
        },
    )
    print(response.choices[0].message.content)
except SecurityBlockError as exc:
    print("Blocked by AgentID:", exc.reason)

By default, official AgentID SDK integrations inherit enable_sdk_pii_masking from the dashboard/runtime config. Use pii_masking=True only when you want to force local masking on even if the dashboard policy is off.

When SDK-side masking is active, caller-facing output stays masked by default. If the model returns an AgentID placeholder from this request, such as <PERSON_1> or <EMAIL_1>, the wrapper returns that placeholder to your app and stores masked telemetry. If an internal route intentionally needs the older placeholder round-trip behavior, opt in explicitly:

secured = agent.wrap_openai(
    openai,
    system_id=os.environ["AGENTID_SYSTEM_ID"],
    deanonymize_output_for_client=True,
)

Do not enable this for data-minimization deployments unless the application is allowed to receive original values back in the model response path.

Constructor-level OpenAI telemetry is copied to guard, local policy telemetry, and final ingest. Per-call agentid_telemetry overrides or extends it and is removed before the OpenAI provider request is sent.

Scope note: AgentID compliance/risk controls apply to the specific SDK-wrapped LLM calls (guard(), wrap_openai(), LangChain callback-wrapped flows). They do not automatically classify unrelated code paths in your whole monolithic application.

Enterprise rollout rule: the main production risk is usually not the classifier itself. It is an unwrapped provider path that still sends prompt/history outside wrap_openai(), LangChain callbacks, or an explicit guard -> provider -> log flow. If the application uses another OpenAI helper, responses.create, Assistants, or a custom provider client, integrate that path explicitly or treat it as a coverage gap.

Before calling a deployment enterprise-ready, verify all of these:

1. Every production LLM callsite uses wrap_openai(), a LangChain callback-wrapped model, or an explicit manual guard -> provider -> log flow.
2. No parallel raw provider call exists in the same request path.
3. Chat and agent routes protect the full message history sent to the provider, not just the latest user text.
4. Manual integrations mark history as protected only when the exact provider payload was rewritten before dispatch.

Current wrapper scope note: wrap_openai() preflight currently derives its guard() input from the last user message text. It does not reconstruct the exact full multi-turn history into the guard request. If you need preflight evaluation over the assembled history today, use an explicit manual guard -> provider -> log flow for that route.

Manual Python routes can protect the exact message history with protect_message_history() before provider dispatch:

from agentid import protect_chat_state, protect_message_history

protected = protect_message_history(body["messages"])

response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=protected.messages,
)

For custom Python/FastAPI/Django chat routes, protect_chat_state() builds separate transcript views for UI, provider dispatch, and persistence:

from agentid import protect_chat_state

protected = protect_chat_state(
    body["messages"],
    transcript={
        "ui": "masked",
        "provider": "masked",
        "persistence": "masked",
    },
)

response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=protected.provider_messages,
)

return {
    "messages": protected.ui_messages,
    "agentid": {
        "transcript": protected.transcript,
        "transformed": protected.transformed,
    },
}

If the product intentionally shows raw user prompts locally, keep that explicit while protecting provider and database paths:

protected = protect_chat_state(
    body["messages"],
    transcript={"ui": "raw", "provider": "masked", "persistence": "masked"},
)

The placeholder mapping is local to the request/session. Do not persist it unless your app has an explicit encrypted vault design.

Use protected.transformed_text_parts_count and protected.text_parts_count in manual telemetry metadata when you do not use wrap_openai().

LangChain Integration

pip install agentid-sdk openai langchain langchain-openai

import os
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from agentid import (
    AgentID,
    AgentIDCallbackHandler,
    create_agentid_correlation_id,
    create_agentid_telemetry_context,
)

agent = AgentID()
workflow_run_id = create_agentid_correlation_id()
handler = AgentIDCallbackHandler(
    agent,
    system_id=os.environ["AGENTID_SYSTEM_ID"],
    telemetry=create_agentid_telemetry_context(
        {
            "workflowRunId": workflow_run_id,
            "workflowStepName": "answer_question",
            "toolName": "langchain.chat",
            "toolTargetType": "conversation",
            "eventCategory": "ai",
            "eventSubtype": "answer_generated",
        }
    ),
)

prompt = PromptTemplate.from_template("Answer in one sentence: {question}")
model = ChatOpenAI(model="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"])
chain = prompt | model | StrOutputParser()

result = chain.invoke(
    {"question": "What is the capital of the Czech Republic?"},
    config={"callbacks": [handler]},
)
print(result)

Constructor-level LangChain telemetry is copied to the guard request, local policy telemetry, and final ingest log. You can override or extend it per invocation with LangChain metadata: config={"metadata": {"agentid_telemetry": {"workflowStepName": "..."}}}.

Raw Ingest API (Telemetry Only)

import os
from agentid import AgentID

agent = AgentID()
agent.log(
    system_id=os.environ["AGENTID_SYSTEM_ID"],
    event_type="complete",
    severity="info",
    model="gpt-4o-mini",
    input="Raw telemetry prompt",
    output='{"ok": true}',
    metadata={"agent_role": "batch-worker", "channel": "manual_ingest"},
)

Agent workflow and tool events

Use log_operation() when an agent calls tools or performs operational work outside the wrapped LLM call. Reuse the same workflow_run_id across steps.

import os
from agentid import (
    AgentID,
    create_agentid_correlation_id,
    create_agentid_telemetry_context,
)

agent = AgentID()
workflow_run_id = create_agentid_correlation_id()

agent.log_operation(
    os.environ["AGENTID_SYSTEM_ID"],
    telemetry=create_agentid_telemetry_context(
        {
            "workflowRunId": workflow_run_id,
            "workflowStepName": "screen_candidate",
            "toolName": "hr.cv_screen",
            "toolTargetType": "candidate",
        }
    ),
    event_category="tool",
    event_status="completed",
)

agent.log_operation(
    os.environ["AGENTID_SYSTEM_ID"],
    telemetry=create_agentid_telemetry_context(
        {
            "workflowRunId": workflow_run_id,
            "workflowStepName": "send_followup",
            "toolName": "email.send",
            "toolTargetType": "email",
        }
    ),
    event_category="delivery",
    event_status="completed",
)

Tool, delivery, inbox, workflow, guard, and operational events are logged as separate audit rows. They are grouped in the dashboard by workflow_run_id. Do not reuse one client_event_id for the whole workflow; use workflow_run_id for grouping and let each event keep its own idempotency key.

Activity Rows vs Workflow Timeline

• prompt/guard checks remain visible as standalone Activity rows with prompt detail
• workflow summary rows open the grouped timeline for tools, delivery, inbox, workflow lifecycle, guard checks, and LLM steps
• the workflow timeline is operational context; the standalone prompt row is the forensic prompt inspection surface
• non-model workflow/tool/delivery rows show Model: Not applicable and are not spend-bearing unless model/cost metadata is explicitly present

For full agent runs, prefer the workflow trail helper so each step gets a shared workflow_step_id, plus automatic started/completed/failed rows:

import os
from agentid import (
    AgentID,
    create_agentid_correlation_id,
    create_agentid_telemetry_context,
    create_agentid_workflow_trail,
)

agent = AgentID(api_key=os.environ["AGENTID_API_KEY"])
workflow_run_id = create_agentid_correlation_id()

trail = create_agentid_workflow_trail(
    agent=agent,
    system_id=os.environ["AGENTID_SYSTEM_ID"],
    telemetry=create_agentid_telemetry_context(
        {
            "workflowRunId": workflow_run_id,
            "workflowName": "Candidate intake",
        }
    ),
)

trail.run_step(
    {
        "telemetry": create_agentid_telemetry_context(
            {
                "workflowStepName": "screen_candidate",
                "toolName": "hr.cv_screen",
                "toolTargetType": "candidate",
                "eventCategory": "tool",
            }
        )
    },
    lambda: screen_candidate(),
    complete={"metadata": {"result_count": 4}},
)

6. Advanced Configuration

Custom identity / role metadata

Use user_id for actor identity and metadata for additional context (for example agent_role, environment, trace IDs).

verdict = agent.guard(
    input="Process user request",
    system_id=system_id,
    user_id="service:billing-agent",
)
agent.log(
    system_id=system_id,
    input="Process user request",
    output="Done",
    model="gpt-4o-mini",
    metadata={"agent_role": "billing-agent", "environment": "prod"},
)

Timeouts

agent = AgentID(
    guard_timeout_s=10.0,
    ingest_timeout_s=10.0,
    strict_mode=True,  # fail-closed on connectivity/timeouts
)

Optional client-side fast fail

agent = AgentID(
    failure_mode="fail_close",
    client_fast_fail=True,  # opt-in local preflight before /guard
)

Error Handling & Strict Mode

By default, AgentID is designed to keep your application running if the AgentID API has a timeout or is temporarily unreachable.

• guard() returns a verdict (allowed, reason); handle deny paths explicitly.
• Wrapped OpenAI/LangChain flows raise SecurityBlockError when a prompt is blocked.
• Backend /guard is the default authority for prompt injection, DB access, code execution, and PII leakage in SDK-wrapped flows.
• client_fast_fail is optional and disabled by default. Enable it only when you explicitly want local preflight before the backend call.
• If backend guard is unreachable and the effective failure mode is fail_close, wrapped OpenAI/LangChain flows can run local fallback enforcement. Local hits still block; otherwise the request can continue with fallback telemetry attached.
• If strict_mode is not explicitly set in SDK code, runtime behavior follows the system configuration from AgentID (strict_security_mode / failure_mode).
• Ingest retries transient failures (5xx/429) and logs warnings if persistence fails.

Event Identity Model

For consistent lifecycle correlation in Activity/Prompts, use this model:

• client_event_id: external correlation/idempotency ID for one SDK-guarded provider call or one operation event.
• guard_event_id: ID of the preflight guard event returned by guard().
• event_id on log(): idempotency key for ingest. In the Python SDK it is canonicalized to client_event_id for stable one-row lifecycle updates.

SDK behavior:

• guard() sends client_event_id and returns canonical client_event_id + guard_event_id.
• log() sends:
  • event_id = canonical client_event_id
  • metadata.client_event_id
  • metadata.guard_event_id (when available from wrappers/callbacks)
  • x-correlation-id = client_event_id
• after a successful primary ingest, SDK wrappers can call /ingest/finalize with the same client_event_id to attach sdk_ingest_ms
• SDK requests include x-agentid-sdk-version for telemetry/version diagnostics.

This keeps Guard + Complete linked under one correlation key while preserving internal event linkage in the dashboard. For multi-step agent workflows, use workflow_run_id as the shared grouping key and keep client_event_id event-specific.

SDK Timing Telemetry

SDK-managed metadata can include:

• sdk_config_fetch_ms: capability/config fetch time before dispatch.
• sdk_local_scan_ms: optional local enforcement time (client_fast_fail or fail-close fallback path).
• sdk_guard_ms: backend /guard round-trip time observed by the SDK wrapper.
• sdk_ingest_ms: post-ingest transport timing finalized by the SDK through /ingest/finalize after a successful primary /ingest.

Policy-Pack Runtime Telemetry

When the backend uses compiled policy packs, runtime metadata includes:

• policy_pack_version: active compiled artifact version.
• policy_pack_fallback: true means fallback detector path was used.
• policy_pack_details: optional diagnostic detail for fallback/decision trace.

Latency interpretation:

• Activity Latency (ms) maps to synchronous processing (processing_time_ms).
• Async AI audit time is separate (ai_audit_duration_ms) and can be higher.
• First request after warm-up boundaries can be slower than steady-state requests.

Secret and PII Masking Edge Cases

SDK-side masking and the backend scanner include regression coverage for common boundary failures:

• multiline PEM, certificate, and PGP private key blocks
• natural-language password disclosures such as my Password is Passwordk123
• environment-style assignments such as DB_PASSWORD=...
• secret values with suffix punctuation such as #
• high-entropy base64-like values with = / == padding
• security-question answers where the value appears after answer is, is, or localized equivalents

When local masking is enabled, these values are replaced before provider dispatch and before AgentID ingest. Placeholder mappings stay local to the SDK for reversible deanonymization.

Monorepo QA Commands (Maintainers)

If you are validating runtime in the AgentID monorepo:

npm run qa:policy-pack-bootstrap -- --base-url=http://127.0.0.1:3000/api/v1 --system-id=<SYSTEM_UUID>
npm run bench:policy-pack-hotpath

PowerShell diagnostics:

powershell -ExecutionPolicy Bypass -File .\scripts\qa\run-guard-diagnostic.ps1 -BaseUrl http://127.0.0.1:3000/api/v1 -ApiKey $env:AGENTID_API_KEY -SystemId $env:AGENTID_SYSTEM_ID -SkipBenchmark
powershell -ExecutionPolicy Bypass -File .\scripts\qa\run-ai-label-audit-check.ps1 -BaseUrl http://127.0.0.1:3000/api/v1 -ApiKey $env:AGENTID_API_KEY -SystemId $env:AGENTID_SYSTEM_ID -Model gpt-4o-mini

7. Security & Compliance

• Optional local-first reversible PII masking via PIIManager and pii_masking=True.
• Backend /guard remains the primary enforcement authority by default.
• client_fast_fail is opt-in; local enforcement is otherwise reserved for fail-close outage fallback.
• Telemetry logging is async/fire-and-forget to minimize app latency, with SDK timing breakdowns finalized on the lifecycle row.
• Designed for server, serverless, and background-worker runtimes.
• Supports compliance workflows requiring complete prompt/output traceability.

8. Support

• Dashboard: https://app.getagentid.com
• Repository: https://github.com/ondrejsukac-rgb/agentid/tree/main/python-sdk
• Issues: https://github.com/ondrejsukac-rgb/agentid/issues

9. Publishing Notes (PyPI)

PyPI renders this README.md as package long description.

Before publishing from the monorepo, run the root release gate:

npm run audit:all
npm run qa:production-gate

Python package tests should also be run from python-sdk/ when changing Python runtime code. The PyPI long description is this file via pyproject.toml.

setup.py projects

from setuptools import setup

with open("README.md", "r", encoding="utf-8") as fh:
    long_description = fh.read()

setup(
    name="agentid-sdk",
    long_description=long_description,
    long_description_content_type="text/markdown",
)

pyproject.toml projects

readme = { file = "README.md", content-type = "text/markdown" }