Datagusto SDK for LangChain / LangGraph / DeepAgents

datagusto-sdk is a LangChain AgentMiddleware that integrates Datagusto safety validation into any agent built with langchain.agents.create_agent or deepagents.create_deep_agent.

It uses the Datagusto public API for tool registration, alignment, on_start / on_end validation, assigned-guardrail evaluation, and prompt / final-response inspection.

Why this exists / what it fixes

Earlier integrations (including datagusto-sdk 0.1.x) raised an exception from inside the tool node when Datagusto blocked a call. That left the assistant's tool_calls without a matching tool response, which corrupts the transcript and makes the very next model call fail — e.g. OpenAI returns:

400 - An assistant message with 'tool_calls' must be followed by tool messages
responding to each 'tool_call_id'. The following tool_call_ids did not have
response messages: ...

This plugin instead turns every block into a well-formed ToolMessage (status="error", carrying the original tool_call_id). The transcript stays valid, and the model reads the block as a tool result and continues the conversation gracefully instead of crashing.

AIMessage(tool_calls=[delete_file(call_1)])
ToolMessage(call_1, status="error",
            content="[BLOCKED by Datagusto guardrail] Tool 'delete_file' was not executed ...")
AIMessage("I couldn't delete the file because a safety guardrail blocked it.")

Prerequisites

• Python ≥ 3.10
• langchain ≥ 1.3, langchain-core ≥ 1.4 and langgraph ≥ 1.2 (for the AgentMiddleware surface)
• A running Datagusto server with a valid API key
• deepagents ≥ 0.3 only if you use create_deep_agent

Installation

pip install datagusto-sdk
# or, with the DeepAgents extra for the example:
pip install "datagusto-sdk[deepagents]"

The distribution is datagusto-sdk; the import package is datagusto_langgraph.

Quick start

With create_agent

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from datagusto_langgraph import DatagustoSafetyMiddleware

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[
        DatagustoSafetyMiddleware(
            server_url="https://your-datagusto-server.example.com",
            api_key="your-api-key",
            # or omit both and set DATAGUSTO_SAFETY_SERVER_URL / DATAGUSTO_SAFETY_API_KEY
        )
    ],
    checkpointer=InMemorySaver(),  # gives each conversation a stable thread_id
)

agent.invoke(
    {"messages": [{"role": "user", "content": "Delete /etc/hosts"}]},
    config={"configurable": {"thread_id": "thread-1"}},
)

With create_deep_agent

from deepagents import create_deep_agent
from datagusto_langgraph import DatagustoSafetyMiddleware

agent = create_deep_agent(
    tools=tools,
    instructions="You are a careful research assistant.",
    middleware=[DatagustoSafetyMiddleware(agent_type="deepagents")],
)

DeepAgents inserts user middleware into its stack and forwards it to create_agent, so the plugin behaves identically in both.

Runnable versions are in the examples/ directory of the source distribution.

Configuration

All options are constructor keyword arguments. Each one also reads a DATAGUSTO_SAFETY_* environment variable; an explicit constructor argument takes precedence over the env var, which takes precedence over the default.

Argument	Env var	Type	Default	Description
server_url	DATAGUSTO_SAFETY_SERVER_URL	str	required	Datagusto base URL
api_key	DATAGUSTO_SAFETY_API_KEY	str	required	Bearer token
timeout_s	DATAGUSTO_SAFETY_TIMEOUT_S	float	30	Per-request timeout (seconds)
fail_closed	DATAGUSTO_SAFETY_FAIL_CLOSED	bool	true	Block tool calls and prompts when the API errors
verbose	DATAGUSTO_SAFETY_VERBOSE	bool	false	INFO-level logging of every decision
monitor_only	DATAGUSTO_SAFETY_MONITOR_ONLY	bool	false	Log verdicts without enforcing them; also relaxes fail-closed. OR-ed with the platform's per-agent monitoring mode — see Platform monitoring mode
agent_type	DATAGUSTO_SAFETY_AGENT_TYPE	str	"langgraph"	Self-declared framework (AgentProfile selection)
use_similar_sessions	DATAGUSTO_SAFETY_USE_SIMILAR_SESSIONS	bool	false	Reference past similar sessions during alignment
on_warn	DATAGUSTO_SAFETY_ON_WARN	"inject" | "interrupt"	"inject"	Behaviour for an on_start warn verdict
allow_uncategorized_tools	DATAGUSTO_SAFETY_ALLOW_UNCATEGORIZED_TOOLS	bool	true	Downgrade the alignment layer's uncategorized_tool deny to allow when the assigned guardrails allowed the call
include_schema	—	bool	true	Send tool input/output JSON schemas at registration
tool_filter	—	list[str]	None	Allow-list of tool names to register

Booleans accept true/1/yes or false/0/no (case-insensitive).

If server_url/api_key cannot be resolved, the middleware logs an error and disables itself — every prompt, tool call and response passes through untouched, so it never crashes an otherwise-working agent.

How it works

LangChain exposes the agent lifecycle through middleware hooks. This plugin implements both the sync and async variants of two of them:

Datagusto step	Hook	What happens
Tool registration + alignment	wrap_model_call / awrap_model_call	Extract tools from request.tools and the latest user message from request.messages; POST /tools/register (when the catalog hash changes) and POST /sessions/alignment (when the instruction hash changes). The returned session_id is stored for the thread.
Prompt inspection	same call, before the model runs	On the first model call of a turn (the last message is the user's), POST /sessions/inspect with evaluation_point="on_prompt_submit".
Final-response inspection	same call, after the model runs	When the model answers without calling tools, POST /sessions/inspect with evaluation_point="on_response".
Pre-execution checks	wrap_tool_call / awrap_tool_call	POST /sessions/validate and POST /guardrails/evaluate with timing="on_start" before running the tool.
Post-execution checks	same call	The same two calls with timing="on_end" and the serialized output (always an object; non-dict outputs are wrapped as {"result": ...}) after running the tool.

Two layers per tool call

Every tool call is checked by two layers:

• /sessions/validate — the alignment (intent-judge) layer. It is session-scoped, so it is skipped while the thread has no alignment session_id (e.g. alignment failed). A missing session is not an error.
• /guardrails/evaluate — the agent's assigned, dashboard-authored guardrail rules. It is agent-scoped via the API key and always runs. It is the only path that evaluates assigned rules.

The verdicts are merged block-wins. A block is attributed to the assigned rule first (its message is user-authored), then to alignment. With allow_uncategorized_tools on (the default), the alignment layer's uncategorized_tool default-deny — a registered tool its intent-judge didn't predict from the prompt — is downgraded to allow, but only when the assigned guardrails explicitly allowed the call. Real intent violations and assigned-rule blocks are never downgraded.

Verdict handling

Timing	Verdict (merged)	Action
on_start	block	Return a ToolMessage(status="error"); the tool does not run.
on_start	severity: "medium" + an instruction	Per on_warn: "inject" runs the tool and appends the note to its result; "interrupt" pauses via langgraph.types.interrupt for human approval.
on_end	block	Replace the tool output with a ToolMessage(status="error") suppression notice.
on_end	an instruction (not blocked)	Append the directive to the tool result as a [Datagusto safety note].
either	otherwise	Pass through unchanged.

The severity and instruction come from the alignment response, else from the assigned rules' warn actions (warning_message / severity). A warn action with allow_proceed: false counts as a block.

Prompt and final-response inspection

/sessions/inspect runs the organization's detector policies (PII, secrets, ...) over the user's prompt and over the agent's final answer. It needs the thread's alignment session, so it is skipped while there is none, and also when the server predates the endpoint (an unknown-route 404).

• Prompt blocked — the model is not called. The turn ends with an AIMessage that explains the block, and the prompt is removed from the thread (RemoveMessage), so it never reaches the model on a later turn. Re-sending the same prompt is inspected — and blocked — again. An inspection API error blocks the prompt under fail_closed (unless monitor_only).
• Response blocked — the answer is dropped and the model is asked once to write it again without the flagged content. The rewrite instruction exists only in that model request and never enters the thread. If the rewritten answer is blocked too, it is replaced by a [BLOCKED by Datagusto guardrail] The response was withheld message. An inspection API error never blocks a response (a warning is logged).

If you stream with stream_mode="messages", a blocked response's tokens have already been streamed before the inspection runs; only the final state is cleaned up.

With DeepAgents, subagents that run this middleware inspect their task description as a prompt and their result as a final response.

Human-in-the-loop (on_warn="interrupt")

When set, an on_start warn verdict calls interrupt({...}) so a reviewer can approve or deny. This requires a checkpointer and that your application resumes the graph with Command(resume=<decision>). The resume value is interpreted leniently: True / "approve" / {"approved": True} → proceed; anything else → block. If interrupt is unavailable, it falls back to injecting the note.

Fail-closed vs fail-open

fail_closed controls behaviour when the Datagusto API is unreachable or returns a non-2xx status:

• fail_closed=True (default): the tool call (on_start) or its output (on_end) is blocked with an error ToolMessage when either layer errors, and a prompt whose inspection errors is blocked. Safe default for production.
• fail_closed=False: the call proceeds, judged by whichever layer answered. Useful for development when the server may be intermittently available.

monitor_only relaxes fail-closed handling too. The platform's monitoring mode does not (an errored request carries no verdict). Final-response inspection never fails closed.

Platform monitoring mode

Independently of monitor_only, the Datagusto platform has a per-agent monitoring mode (dashboard: agent settings → monitoring). When it is on, the server still evaluates guardrails, records every verdict and detects incidents, but returns monitoring_mode_enabled: true on /sessions/alignment, /sessions/validate and /sessions/inspect responses and expects the client not to enforce them. /guardrails/evaluate does not carry the flag; its verdicts follow the value from validate or the thread's cached value.

The plugin honours that flag: effective monitoring = monitor_only OR monitoring_mode_enabled. Either switch disables enforcement; there is no way for one to override the other. The platform value is resolved per request:

1. monitoring_mode_enabled on the current validate / inspect response (follows a settings change mid-session), then
2. the value most recently reported by any alignment, validate or inspect response for the current thread, then
3. false — servers that omit the field behave exactly as before.

The first time a thread sees the platform flag as true, one logging.WARNING record is emitted on the datagusto.langgraph logger; every skipped verdict is then logged as [MONITOR] ... skipped (platform) (vs. (plugin)) at INFO when verbose is on. Fail-closed handling of API errors is unaffected by the platform flag, because an errored request carries no verdict.

Per-thread sessions

State (alignment session_id plus de-duplication hashes) is keyed by the graph's thread_id, read automatically from the runtime via langgraph.config.get_config(). You do not need to swap a session id on the middleware between turns — each conversation keeps its own Datagusto session as long as you pass a stable thread_id (i.e. use a checkpointer).

Migrating from datagusto-sdk 0.1.x

datagusto-sdk 1.0 is a rewrite and is not compatible with 0.1.x:

# before (0.1.x)
from datagusto_sdk import DatagustoSafetyMiddleware
middleware = DatagustoSafetyMiddleware(verbose=False, include_schema=True)

# after (1.x)
from datagusto_langgraph import DatagustoSafetyMiddleware
middleware = DatagustoSafetyMiddleware(verbose=False, include_schema=True)

• The import package is now datagusto_langgraph.
• Blocks no longer raise. They are returned as ToolMessages, so remove any try/except around agent.invoke(...) / agent.stream(...). To detect a block, look for a ToolMessage with status == "error" whose content starts with [BLOCKED by Datagusto guardrail].
• No manual session juggling. Sessions are tracked per thread_id automatically.
• SERVER_URL / API_KEY env vars are now DATAGUSTO_SAFETY_SERVER_URL / DATAGUSTO_SAFETY_API_KEY (constructor args still work), and all arguments are keyword-only.
• verbose now defaults to false.

Architecture

create_agent / create_deep_agent
  |
  |-- wrap_model_call ----> build_tool_definitions(request.tools) --> POST /tools/register
  |                        extract_latest_user_instruction()       --> POST /sessions/alignment
  |                                                                     (store session_id per thread_id)
  |                        last message is the user's prompt       --> POST /sessions/inspect (on_prompt_submit)
  |                           blocked  --> answer with the reason, remove the prompt  [model not called]
  |                        model answers without tool calls        --> POST /sessions/inspect (on_response)
  |                           blocked  --> ask the model to rewrite once; still blocked --> withhold
  |
  |-- wrap_tool_call -----> POST /sessions/validate + POST /guardrails/evaluate (on_start), merged block-wins
  |                           block                  --> return ToolMessage(status="error")  [tool skipped]
  |                           warn + instruction     --> inject note  | interrupt for approval
  |                           otherwise              --> run tool
  |                         POST /sessions/validate + POST /guardrails/evaluate (on_end), merged block-wins
  |                           block                  --> return ToolMessage(status="error")  [output suppressed]
  |                           injected instruction   --> append [Datagusto safety note]
  |                           otherwise              --> pass result through

License

MIT — see the LICENSE file.