> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cogensec.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python frameworks

> Install extras, the init() entry point, and the zero-config module for each of the eight Python framework adapters.

Each Python adapter runs the four evaluation layers in your process, appends to a hash-linked attestation chain, and reports a session summary. Nothing leaves the process unless you register an exporter or set the streaming environment variables (see [Agentegrity Pro](https://docs.cogensec.com/exporters/agentegrity-pro.md)).

## Install

The base package has no dependencies. Each framework has an extra that installs it:

```bash theme={null}
pip install "agentegrity[claude]"
```

| Extra | Installs |
| - | - |
| `claude` | `claude-agent-sdk>=0.1.0` |
| `langchain` | `langchain-core>=0.3.0` |
| `openai-agents` | `openai-agents>=0.1.0` |
| `crewai` | `crewai>=1.15.0` |
| `google-adk` | `google-adk>=1.18.0` |
| `autogen` | `autogen-agentchat>=0.7.0`, `autogen-core>=0.7.0`, `opentelemetry-sdk>=1.20` |
| `agno` | `agno>=3.1` |
| `bedrock-agents` | `boto3>=1.34`, `strands-agents>=1.0` |

The `crewai` and `google-adk` floors are the first releases with the APIs the token usage code reads (`UsageMetrics.from_provider_dict` and `LlmResponse.model_version`). An older CrewAI installed outside the extra still runs the adapter; it reports lifecycle events and no usage. The `langchain` extra installs `langchain-core` only; install LangGraph yourself if you use it.

Other extras: `crypto` (Ed25519 signing), `llm` (LLM-backed checks through the Anthropic API), `stats` (Wasserstein drift metric), `kms` (KMS-encrypted checkpoints), `otel` (the OpenTelemetry exporter), and `all`. See [Installation](https://docs.cogensec.com/installation.md).

## Two-line start: `init()`

`agentegrity.init()` checks which frameworks are importable, builds one adapter per framework on a shared `AgentegrityClient` and profile, and subscribes CrewAI to its global event bus. It returns an `AgentegrityRuntime`.

```python theme={null}
import agentegrity

runtime = agentegrity.init()
graph = runtime.instrument(my_graph)   # LangChain / LangGraph / Agno / Strands / Google ADK
graph.invoke(state)
print(runtime.report())                # {adapter name: session summary}
agentegrity.shutdown()                 # ends every session, sends on_session_end
```

`init()` takes keyword arguments only:

**name** `str` (default: agent)

Name for the default `AgentProfile`. Ignored when `profile` is passed.

**frameworks** `list[str] | None` (default: None)

Adapter names to build instead of detecting them: `claude`, `langchain`, `openai_agents`, `crewai`, `google_adk`, `autogen`, `agno`, `bedrock_agents`. An unknown name raises `ValueError`.

**enforce** `bool` (default: False)

Forwarded to every adapter. The two observation-only surfaces (AutoGen and the Bedrock boto3 path) warn.

**policy\_set** `str` (default: enterprise-default)

Governance policy set for the shared client.

**profile** `AgentProfile | None` (default: None)

Profile to use instead of a default one.

Detection is by import: `claude_agent_sdk`, `langchain_core`, `agents`, `crewai`, `google.adk`, `autogen_agentchat` or `autogen_core`, `agno`, and `strands`. The Bedrock adapter is detected by `strands` only, so a boto3-only project passes `frameworks=["bedrock_agents"]`. When a framework is importable but its adapter's own dependencies are missing, the adapter is skipped and the reason is kept in `runtime.skipped`.

`runtime.instrument(obj)` dispatches on the module of the object's class: LangGraph graphs go to `instrument_graph`, other LangChain runnables to `instrument_chain`, Agno teams to `instrument_team`, Agno agents to `instrument`, Strands agents to `instrument_strands`, and Google ADK agents to `instrument`. It raises `TypeError` for an object it cannot place, and `RuntimeError` when the object's adapter was not built. CrewAI needs no call: `init()` already subscribed it.

Hook-style frameworks and AutoGen are wired from the adapter itself:

```python theme={null}
options = ClaudeAgentOptions(hooks=runtime.adapters["claude"].create_hooks())
await Runner.run(agent, input="...", hooks=runtime.adapters["openai_agents"].create_run_hooks())
runtime.adapters["autogen"].instrument()   # installs a global OTel TracerProvider
```

AutoGen is not wired by `init()` because its `instrument()` replaces the global OpenTelemetry tracer provider, which should be your choice.

A second `init()` returns the existing runtime with a `RuntimeWarning`. Call `agentegrity.shutdown()` first to reconfigure. `shutdown()` is safe to call when `init()` never ran. `agentegrity.detect_frameworks()` returns the detected adapter names without building anything.

## Per-framework modules

Each framework also has a zero-config module, imported as `agentegrity.<framework>`. On first use the module builds a process-global adapter with `AgentProfile.default()` and the default client, and every later call reuses it. Each module exports the same four helpers:

| Function | What it does |
| - | - |
| `report()` | The module-global adapter's session summary. Returns a well-formed empty summary if the adapter was never built. |
| `register_exporter(exporter)` | Registers a [`SessionExporter`](https://docs.cogensec.com/exporters/overview) on the module-global adapter, building it if needed. |
| `adapter()` | The module-global adapter itself, built on first call. |
| `reset()` | Discards the module-global adapter. It does not end the session. |

> [!WARNING]
>
> A module-global adapter does not end its session on its own. Call `adapter().close()` when the run finishes so exporters receive `on_session_end` and the cortical layer learns the session. `close()` is safe to call twice. Call it before `reset()`, because `reset()` drops the adapter without closing it.

### Claude Agent SDK

```python theme={null}
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
from agentegrity.claude import adapter, hooks, report

async with ClaudeSDKClient(options=ClaudeAgentOptions(hooks=hooks())) as sdk:
    await sdk.query("Summarize the latest LLM safety papers")
    async for message in sdk.receive_response():
        adapter().observe(message)   # optional: exact token totals
adapter().close()
print(report())
```

**Hooks into:** `hooks()` returns `HookMatcher` lists for eight SDK hook events: `PreToolUse`, `PostToolUse`, `PostToolUseFailure`, `UserPromptSubmit`, `Stop`, `SubagentStart`, `SubagentStop` and `PreCompact`. Each maps to the canonical event of the same name.

**Enforcement:** with `enforce=True`, the `PreToolUse` hook returns a `permissionDecision: "deny"` payload when the score's action is `block`, or `escalate` without an approving handler. The SDK refuses the tool call.

**Topology:** none. The Claude Agent SDK has no multi-agent primitive, so the adapter never declares one; subagents appear as `subagent_start` and `subagent_stop` events.

**Token usage:** read from the session transcript the hooks name, with no code change. Passing each message to `adapter().observe(message)` replaces that estimate with `ResultMessage.model_usage`, the SDK's per-model totals, subagents included.

### LangChain and LangGraph

```python theme={null}
from agentegrity.langchain import instrument_chain, instrument_graph, report

chain = instrument_chain(my_chain)
chain.invoke({"input": "..."})

graph = instrument_graph(my_compiled_graph)
graph.invoke(state)
print(report())
```

For an agent built with `create_agent`, or a LangGraph `ToolNode`, gate tool calls with middleware or tool wrappers so a denied call reaches the model and the run continues. Take both from one adapter, and keep its callback handler on the graph for prompts and the end of the run:

```python theme={null}
from langchain.agents import create_agent
from langgraph.prebuilt import ToolNode
from agentegrity import AgentegrityClient, AgentProfile

ad = AgentegrityClient().create_adapter(
    "langchain", profile=AgentProfile.default(name="my-agent"), enforce=True
)
agent = ad.instrument_graph(create_agent(model, tools, middleware=[ad.middleware()]))
tool_node = ToolNode(tools, **ad.tool_wrappers())
```

Put the Agentegrity middleware last in `middleware=[...]`: it evaluates the call it receives, so a middleware after it could change the call after the check. `middleware()` needs `langchain` 1.0 or later and `tool_wrappers()` needs `langgraph-prebuilt` 1.0.2 or later. The module functions `middleware()` and `tool_wrappers()` use the module's default adapter, which does not enforce; with any argument they build a new adapter each call. Use the returned object. `instrument_chain()` wraps the runnable with `with_config(callbacks=[handler])` and leaves the original unchanged; a legacy chain without `with_config` gets the handler appended to its `callbacks` attribute instead. `callback_handler()` returns the raw `BaseCallbackHandler` if you register callbacks yourself.

**Hooks into:** a LangChain callback handler. A top-level `on_chain_start` becomes `user_prompt_submit`, a nested chain start (a graph node) becomes `subagent_start`, a top-level `on_chain_end` becomes `stop`, and `on_tool_start`, `on_tool_end` and `on_tool_error` become the three tool events.

**Enforcement:** with `middleware()` or `tool_wrappers()`, a denied call returns a `ToolMessage` with `status="error"` and the [denial code](#denials) instead of running, and the agent continues. With `on_deny="raise"` it raises `agentegrity.IntegrityViolationError` instead. With only the callback handler, LangChain calls `on_tool_start` before it runs the tool body and the handler sets `raise_error`, so a denied call raises `IntegrityViolationError` and the `invoke` call ends: a callback cannot hand the model a result. A call that the same adapter's middleware or tool wrappers are running is left to them (matched by tool call id), so it is evaluated and recorded once; every other call the callback handler still gates.

**Topology:** `instrument_graph()` reads `graph.get_graph().nodes`. A node named `supervisor`, `supervisor_agent` or `orchestrator` gives a `HIERARCHICAL_DAG` with that node as supervisor and the rest as workers; otherwise every node is a peer in a `PEER_TO_PEER` topology. If introspection fails, the graph is instrumented as a single agent. `instrument_chain()` declares no topology.

**Token usage:** `on_llm_end` reads each generation's `usage_metadata`, keyed by run id so a handler attached twice counts a call once. The model comes from `response_metadata`, or from the chat model's start metadata.

### OpenAI Agents SDK

```python theme={null}
from agents import Agent, Runner
from agentegrity.openai_agents import instrument, run_hooks, report

agent = instrument(Agent(...))
await Runner.run(agent, input="...", hooks=run_hooks())
print(report())
```

`instrument()` needs `openai-agents` 0.8 or later. The hooks skip calls whose tool carries the same adapter's guardrail, so take both from one adapter. To enforce:

```python theme={null}
from agentegrity import AgentegrityClient, AgentProfile

ad = AgentegrityClient().create_adapter(
    "openai_agents", profile=AgentProfile.default(name="my-agent"), enforce=True
)
await Runner.run(ad.instrument(Agent(...)), input="...", hooks=ad.create_run_hooks())
```

The module functions use the default adapter, which does not enforce; with any argument they build a new adapter each call, so `instrument(agent, enforce=True)` and `run_hooks(enforce=True)` would be two adapters that both evaluate every call.

**Hooks into:** a `RunHooks` subclass. `on_agent_start` becomes `user_prompt_submit`, `on_tool_start` becomes `pre_tool_use`, `on_tool_end` becomes `post_tool_use`, `on_handoff` becomes `subagent_start`, and `on_agent_end` becomes `stop`. `pre_tool_use` carries the call's arguments, parsed from the JSON string the SDK passes on `ToolContext`. `instrument()` adds a tool input guardrail to every function tool the agent can reach: its own tools, the tools of agents used through `Agent.as_tool()`, and those of agents in `handoffs` (for a `handoff(agent)` object, only on SDK versions that keep a reference to the agent, through the private `_agent_ref`).

**Enforcement:** a function tool carrying the guardrail returns the [denial code](#denials) to the model in place of its result (`reject_content`), and the run continues; with `on_deny="raise"`, `Runner.run` ends with `agentegrity.IntegrityViolationError`. Tools without the guardrail, such as MCP tools, which the SDK builds on every turn, are gated in `on_tool_start`, which can only raise. Reaching inside `Agent.as_tool()` relies on a private SDK attribute (`_agent_instance`); `instrument()` warns if an SDK version no longer has it. Tools that run on OpenAI's side (web search, file search) and handoffs cannot be blocked, and shell, local shell, computer and apply\_patch tools are evaluated by name only.

**Topology:** `PEER_TO_PEER`, built incrementally. The starting agent seeds a one-member topology and each handoff target is added as a peer, emitting `topology_change`.

**Token usage:** `on_llm_end` counts each `response.usage`, keyed by `response_id`, with the agent's model. An agent run as a tool shares its parent's `Usage` object but fires no hooks, so whatever the shared total holds beyond the counted calls is recorded with model `unknown`.

### CrewAI

```python theme={null}
from agentegrity.crewai import adapter, instrument, report

instrument(crew)        # or instrument() with no crew
crew.kickoff(inputs={"topic": "llms"})
adapter().close()
print(report())
```

**Hooks into:** the global `crewai_event_bus` and a before-tool-call hook. The hook turns each tool call into `pre_tool_use`, with its arguments; `CrewKickoffStartedEvent` becomes `user_prompt_submit`, `CrewKickoffCompletedEvent` becomes `stop`, `ToolUsageFinishedEvent` and `ToolUsageErrorEvent` become `post_tool_use` and `post_tool_use_failure`, `TaskStartedEvent` becomes `task_started`, and `AgentExecutionStartedEvent` / `AgentExecutionCompletedEvent` become `subagent_start` / `subagent_stop`. The bus handlers see every crew in the process whether or not you pass a crew; the tool hook evaluates the crew you pass and its copies (`kickoff_for_each`, `train` and `test` run copies), matched by `crew.key`, or every crew when you pass none. A different crew with the same agents and tasks has the same key and is evaluated too. `instrument()` returns the adapter, and its `close()` removes the hook.

**Enforcement:** CrewAI skips a tool when a before-tool-call hook returns `False`. Under `enforce=True` the hook returns `False` for a denied call, the tool never runs, the agent receives "Tool execution blocked by hook" as the result, and the crew continues. The hook runs synchronously; the event bus runs its handlers on a thread pool, which is why tool calls are not evaluated from `ToolUsageStartedEvent`.

**Topology:** only when you pass the crew. A sequential process gives `HUB_SPOKE`, with the first agent as leader by convention; a hierarchical process gives `HIERARCHICAL_DAG`. Members are identified by agent role.

**Token usage:** `LLMCallCompletedEvent`, one entry per `call_id`, normalized with CrewAI's `UsageMetrics.from_provider_dict` (crewai 1.15 and later).

> [!NOTE]
>
> Call `instrument()` once per process. A second call registers the handlers on the bus again and every event is evaluated twice. After `agentegrity.init()`, CrewAI is already subscribed.

### Google ADK

```python theme={null}
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from agentegrity.google_adk import instrument, report

runner = instrument(InMemoryRunner(agent=LlmAgent(name="my-agent", ...)))
print(report())
```

Pass the `Runner`, or an `App` before you build its `Runner`, so `instrument()` also registers an ADK plugin. The plugin gates every tool call in the app, including tool nodes in a workflow graph. Passing a bare agent still works and covers that agent's own tools.

**Hooks into:** ADK agent callbacks, placed ahead of any callbacks already set, on the agent and on every agent reachable through `sub_agents` or an `AgentTool`. `before_agent_callback` becomes `user_prompt_submit` on the instrumented agent and `subagent_start` on any other; `after_agent_callback` becomes `stop` or `subagent_stop`; `before_tool_callback` and `after_tool_callback` become the tool events. ADK uses the first value a callback returns. These callbacks return nothing unless enforcement denies a tool call, so otherwise your own callbacks still run and decide. Instrumenting an agent again is a no-op. ADK does not run after-agent callbacks once it has transferred to another agent, so a run that ends in a transfer reports no `stop`.

**Enforcement:** ADK skips a tool when a before-tool callback returns a dict and gives that dict to the model as the tool's result. Under `enforce=True` a denied call returns `{"error": <denial code>}`, so the tool never runs, the model sees the denial, and the run continues. Tool nodes in an ADK workflow graph run only plugin callbacks, so they are gated only when `instrument()` was given the `Runner` or `App`. A call the plugin handles is not evaluated again by the agent callbacks.

**Topology:** an agent with `sub_agents` (`SequentialAgent`, `ParallelAgent`, `LoopAgent`) gives a `HIERARCHICAL_DAG` with the workflow agent as supervisor. A plain agent stays single-agent.

**Token usage:** `after_model_callback`, final responses only, since streamed partials repeat the usage. Thinking tokens are added to output and tool-use prompt tokens to input.

### AutoGen

```python theme={null}
from autogen_agentchat.teams import RoundRobinGroupChat
from agentegrity.autogen import instrument, report

instrument()
team = RoundRobinGroupChat([agent1, agent2])
await team.run(task="...")
print(report())
```

**Hooks into:** OpenTelemetry. AutoGen has no callback API; its `invoke_agent` and `execute_tool` spans go through the global tracer. `instrument()` builds a `TracerProvider` with an Agentegrity `SpanProcessor` and installs it globally, logging a warning if it replaces a provider you set. Pass `set_global=False` to skip the install, or, if you manage your own provider, skip `instrument()` and call `provider.add_span_processor(adapter().span_processor())`. A root `invoke_agent` span becomes `user_prompt_submit` / `stop`, a nested one `subagent_start` / `subagent_stop`, and `execute_tool` becomes `pre_tool_use` and then `post_tool_use` or `post_tool_use_failure` by span status. `instrument()` returns the adapter.

**Enforcement:** observation only. A span processor cannot deny a call. The adapter warns at construction when `enforce=True`. `execute_tool` spans carry the tool name and call id but not the arguments or result.

**Topology:** `GROUP_CHAT`, seeded from the root `invoke_agent` span and grown by nested spans through `topology_change`.

**Token usage:** AutoGen logs `LLMCallEvent` and `LLMStreamEndEvent` to the `autogen_core.events` logger. The adapter lowers that logger to INFO behind one shared filter that records usage and passes on only the records the logger emitted before, so your log output does not change. The level is restored when the last adapter closes.

### Agno

```python theme={null}
from agno.agent import Agent
from agentegrity.agno import instrument, instrument_team, report

agent = instrument(Agent(...))
agent.run("...")

team = instrument_team(team)
team.run("...")
print(report())
```

**Hooks into:** Agno 3.1+ `pre_hooks`, `post_hooks` and `tool_hooks`, appended to any hooks already set. The pre-hook becomes `user_prompt_submit` (or `subagent_start` on a team member), the post-hook becomes `stop` (or `subagent_stop`), and the tool hook wraps every tool call. Agno re-applies `tool_hooks` to every tool at run setup, so tools added after `instrument()` are covered. Under `arun()` the tool hook awaits the gate and the tool on Agno's event loop.

**Enforcement:** under `enforce=True` the tool hook evaluates `pre_tool_use` before the tool runs and, on a deny decision, raises `agno.exceptions.StopAgentRun`, the exception family Agno re-raises to halt the run.

**Topology:** `instrument_team()` declares `HUB_SPOKE` with the team as leader and each member from `team.members` as a member. When `members` is a callable, only the leader is declared and members' `subagent_*` events are not attached; their tool calls are still captured by the leader's tool hook.

**Token usage:** `run_output.metrics` in the post-hook, per model from `metrics.details`, re-read at every stop and at close because Agno merges memory and summary model calls into the metrics after post-hooks run.

### AWS Bedrock Agents

```python theme={null}
from agentegrity.bedrock_agents import instrument_strands, report, wrap_client

# Strands SDK
agent = instrument_strands(agent)
agent("...")

# boto3 bedrock-agent-runtime
import boto3
client = wrap_client(boto3.client("bedrock-agent-runtime"))
response = client.invoke_agent(agentId="...", agentAliasId="...", sessionId="...", inputText="...")
for event in response["completion"]:
    ...
print(report())
```

**Hooks into:** two surfaces on one adapter. `instrument_strands()` registers a hook provider for `BeforeInvocationEvent`, `AfterInvocationEvent`, `BeforeToolCallEvent` and `AfterToolCallEvent`. `wrap_client()` replaces the client's `invoke_agent` with a version that sets `enableTrace=True` (unless you pass `force_trace=False`) and maps trace parts onto canonical events as you consume the stream. `chunk` and other events pass through unchanged.

**Enforcement:** on Strands, a deny decision sets `event.cancel_tool` so Strands does not run the tool. If a Strands version refuses the write, the adapter raises `IntegrityViolationError` instead. The boto3 path is observation only, because traces arrive after the tool ran; `wrap_client()` warns when the adapter has `enforce=True`.

**Topology:** `HUB_SPOKE`, seeded with the Strands agent or the `agentId` as supervisor. On the boto3 path, collaborators found in `agentCollaboratorInvocationInput` traces are added through `topology_change`.

**Token usage:** on Strands, the invocation's usage from `AfterInvocationEvent`. On boto3, `modelInvocationOutput.metadata.usage` from orchestration, pre- and post-processing and routing traces, keyed by trace id; most traces do not name the model, so it falls back to the profile's `model_id`.

## Denials

Under `enforce=True` a denied tool call never runs. What happens next depends on `on_deny`, an argument of every adapter class and of `AgentegrityClient.create_adapter()`:

* `"continue"` (the default) hands the model the denial in place of the tool's result, and the run goes on. The model can try something else, and Phase 1 recovery has a live run to work with.
* `"raise"` ends the run with `agentegrity.IntegrityViolationError`.

| Framework | `continue` | `raise` |
| - | - | - |
| Claude Agent SDK, CrewAI, Google ADK | Always | Refused: an exception from their tool hook does not stop the run |
| AutoGen, Bedrock `wrap_client()` | Observation only: nothing is denied | Refused (AutoGen) |
| LangChain `middleware()` / `tool_wrappers()`, OpenAI `instrument()`, Strands | Yes | Yes |
| LangChain callback handler, OpenAI run hooks (tools without the guardrail), Agno | Not possible: the hook cannot return a result | Always |

The text the model receives is fixed:

```text theme={null}
agentegrity.denied:block: Agentegrity denied this tool call, so it did not run.
```

The code after `agentegrity.denied:` is `block`, `escalate` (no approval) or `error` (the call could not be evaluated). It never names the detector, the rule or the score, so an attacker who probes denials learns only that a call was denied. The score and the rules that fired are in the attestation chain and the session events.

## Explicit configuration

Every entry point (`hooks()`, `instrument_chain()`, `instrument_graph()`, `callback_handler()`, `run_hooks()`, `instrument()`, `instrument_team()`, `instrument_strands()`, `wrap_client()`) accepts three keyword arguments:

**profile** `AgentProfile`

The profile to evaluate against instead of `AgentProfile.default()`.

**client** `AgentegrityClient`

A client with custom thresholds, weights or policy set. `AgentegrityClient(prefer_llm=True)` adds the Claude-backed classifiers at every tool-call gate.

**enforce** `bool` (default: False)

Deny tool calls on adapters with a blocking surface.

Passing any of them builds a new adapter instead of using the module global, so the module's `report()`, `register_exporter()` and `adapter()` do not see it. To keep a handle on that adapter, create it through the client:

```python theme={null}
from agentegrity import AgentegrityClient
from claude_agent_sdk import ClaudeAgentOptions

client = AgentegrityClient(policy_set="enterprise-default", coherence_threshold=0.70)
profile = client.create_profile(name="research-agent", risk_tier="high")
adapter = client.create_adapter("claude", profile=profile, enforce=True)

options = ClaudeAgentOptions(hooks=adapter.create_hooks())
# ... run the agent ...
adapter.close()
print(adapter.get_summary())
```

`create_adapter()` takes the adapter names listed under `init()`. Approval handlers and signing keys are constructor arguments on the adapter classes in `agentegrity.adapters` (for example `ClaudeAdapter(profile=..., enforce=True, approval_handler=...)`); see [Scoring and actions](https://docs.cogensec.com/concepts/scoring-and-actions.md) and [Attestation](https://docs.cogensec.com/concepts/attestation.md). Runnable versions of these flows are in [`examples/`](https://github.com/Cogensec/agentegrity/tree/main/examples).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.