Skip to main content
Drako provides a dedicated wrapper function for each supported framework. All three share a common set of kwargs and can also be called indirectly through the govern() one-liner.

​
with_compliance() — CrewAI

Wraps a CrewAI Crew with a transparent compliance proxy. The returned object exposes the same interface as the original crew — call kickoff() or akickoff() as you normally would.

​
Signature

from drako import with_compliance

crew = with_compliance(
    crew,
    config_path=".drako.yaml",
    auto_audit=True,
    auto_verify=True,
    auto_policy=True,
)

​
Parameters

​
crew
Any
required
A CrewAI Crew instance.
​
config_path
str
default:".drako.yaml"
Path to the .drako.yaml configuration file.
​
auto_audit
bool
default:"true"
Write an audit log entry after every task completes.
​
auto_verify
bool
default:"true"
Call verify_agent_identity for every agent before the first kickoff().
​
auto_policy
bool
default:"true"
Evaluate policy before each task and before each individual tool call.

​
What it does

The wrapper intercepts the full crew lifecycle:
  1. Pre-kickoff — verifies the identity of every agent (when auto_verify=True) and caches their DIDs.
  2. Tool wrapping — replaces each tool’s _run() method with a governed version (when auto_policy=True). The governed _run() runs the following pipeline before calling the original method:
    • Pre-action hooks (/api/v1/hooks/execute)
    • Gate 1: intent fingerprint creation (/api/v1/intent/create)
    • Policy evaluation (/api/v1/trust/evaluate)
    • Gate 2: intent fingerprint verification (/api/v1/intent/verify)
    • FinOps cache lookup (if enabled in config)
    • FinOps model routing (if enabled in config)
  3. Post-task — records an audit log entry via the task callback (when auto_audit=True).
  4. On error — fires on_error hooks.
Unrecognised decisions ("PENDING_APPROVAL", "escalated") pause the agent and return a human-readable message including the approval ID. Blocked decisions ("BLOCKED", "rejected", False) return a [Drako] Action blocked: ... string instead of executing.
The fail_closed behaviour is read automatically from governance.on_backend_unreachable in your .drako.yaml. When set to "block", a backend network error will block the tool. The default ("allow") is fail-open.

​
Example

from crewai import Crew, Agent, Task
from drako import with_compliance

researcher = Agent(role="researcher", goal="...", tools=[search_tool])
task = Task(description="Research recent news", agent=researcher)

crew = Crew(agents=[researcher], tasks=[task])
crew = with_compliance(crew)  # or with_compliance(crew, config_path="/path/to/.drako.yaml")

result = crew.kickoff()

​
with_langgraph_compliance() — LangGraph

Wraps a compiled LangGraph graph by injecting a DrakoCheckpointer that intercepts every state transition.

​
Signature

from drako import with_langgraph_compliance

graph = with_langgraph_compliance(
    graph,
    config_path=".drako.yaml",
    inner_checkpointer=None,
    auto_audit=True,
    auto_policy=True,
)

​
Parameters

​
graph
Any
required
A compiled LangGraph object (result of StateGraph(...).compile()).
​
config_path
str
default:".drako.yaml"
Path to the .drako.yaml configuration file.
​
inner_checkpointer
Any | None
An existing LangGraph checkpointer to delegate put/get calls to after compliance checks. Pass your MemorySaver or SqliteSaver here to combine Drako with built-in checkpointing.
​
auto_audit
bool
default:"true"
Write an audit log entry on every state transition.
​
auto_policy
bool
default:"true"
Evaluate policy on every state transition. The source node name from the checkpoint metadata is used as the action label.

​
What it does

The wrapper returns a _LangGraphProxy that forwards invoke(), ainvoke(), stream(), and astream() to the original graph, injecting the DrakoCheckpointer into the LangGraph config dict on each call. The DrakoCheckpointer fires on every put / aput (state write):
  1. Evaluates policy for the transition node (when auto_policy=True).
  2. Records an audit log entry with the node name and checkpoint timestamp (when auto_audit=True).
  3. Delegates to inner_checkpointer if one was provided.
auto_verify is always False for LangGraph because graph nodes do not have named agent identities by default.

​
Example

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from drako import with_langgraph_compliance

builder = StateGraph(MyState)
builder.add_node("search", search_node)
builder.set_entry_point("search")
builder.add_edge("search", END)

raw_graph = builder.compile()
graph = with_langgraph_compliance(
    raw_graph,
    inner_checkpointer=MemorySaver(),  # optional
)

result = graph.invoke({"messages": ["What is the weather today?"]})

​
with_autogen_compliance() — AutoGen

Adds a silent DrakoObserver agent to an AutoGen GroupChat. The observer audits every message without generating replies.

​
Signature

from drako import with_autogen_compliance

chat = with_autogen_compliance(
    group_chat,
    config_path=".drako.yaml",
    auto_audit=True,
    auto_policy=True,
)

​
Parameters

​
group_chat
Any
required
An AutoGen GroupChat instance.
​
config_path
str
default:".drako.yaml"
Path to the .drako.yaml configuration file.
​
auto_audit
bool
default:"true"
Record an audit log entry for every message exchange.
​
auto_policy
bool
default:"true"
Evaluate policy before processing each message. The sender’s name is used as the action label.

​
What it does

The wrapper returns an _AutoGenGroupChatProxy that:
  1. Agent verification — calls verify_agent_identity for every agent in the chat before run() starts.
  2. Message interception — the DrakoObserver calls on_message() for each exchange, evaluating policy and writing audit log entries.
  3. Transparent proxy — all other attributes are forwarded to the underlying GroupChat.

​
Example

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from drako import with_autogen_compliance

assistant = AssistantAgent("assistant", llm_config={...})
user_proxy = UserProxyAgent("user_proxy", ...)

group_chat = GroupChat(agents=[assistant, user_proxy], messages=[])
group_chat = with_autogen_compliance(group_chat)

manager = GroupChatManager(groupchat=group_chat, llm_config={...})
user_proxy.initiate_chat(manager, message="Summarise last quarter.")

​
Common kwargs

All three wrapper functions accept the following shared parameters:
ParameterTypeDefaultDescription
config_pathstr".drako.yaml"Path to the Drako config file.
auto_auditboolTrueWrite audit log entries.
auto_policyboolTrueEvaluate policy before actions.
auto_verify is also supported by with_compliance() and with_autogen_compliance() but is not applicable to LangGraph (always False).
For most use cases, the defaults are correct. Disable individual checks only when integrating with systems that already provide their own audit or policy layer.

​
Framework examples side-by-side

from drako import with_compliance

crew = with_compliance(crew)
result = crew.kickoff()

# Async
result = await crew.akickoff()

Build docs developers (and LLMs) love

Get started for freeTalk to us