connexum-governance 1.0.0b6

Python client for My Compliance Center AI governance enforcement

connexum-governance

Welcome to the Citadel — My-CC fortress for building Agent Trust. The Python client for My-CC.io AI Agent Trust Citadel.

Python client for My Compliance Center AI governance enforcement.

Thin HTTP client that connects to the Connexum governance API server. The governance engine runs in TypeScript. This client sends governance checks and receives decisions. No engine duplication.

Install

pip install connexum-governance

Quick Start

from connexum_governance import GovernanceClient, AgentIdentity

gov = GovernanceClient(
    api_url="http://localhost:3100",
    license_key="your-license-key",
)

decision = gov.check_action(
    tool="Bash",
    input={"command": "rm -rf /tmp/data"},
    agent=AgentIdentity(name="cleanup-agent"),
)

if decision.denied:
    print(f"Blocked: {decision.reason}")
elif decision.requires_approval:
    print(f"Needs approval: {decision.audit_id}")
else:
    print("Allowed, proceed")

Robotics + Embedded AI

My Compliance Cortex governs the AI brain — the agent runtime that emits tool-call decisions. Governance fires on those decisions before execution.

My-CC enforces policy on the AI agent's tool-call surface. It does NOT directly enforce policy on mechanical actuators, physical sensors, hardware safety interlocks, or real-time control loops. Actuator safety remains the integrating system's responsibility. My-CC provides audit-chain visibility into AI decisions that precede actuator commands; it does not veto those commands at the hardware layer.

Today, ROS 2 Python nodes can use GovernanceClient + any of the 8 framework adapters (LangChain, LlamaIndex, CrewAI, AutoGen, LangGraph, Pydantic AI, Haystack, OpenAI Agents SDK) without additional build. See docs/ROBOTICS_INTEGRATION_PLAN.md for the full spec including the planned ROS 2 native adapter, embedded runtime, air-gap mode, and actuator boundary attestation.

LLM Provider Integrations

Governance wrappers for six LLM providers. Each integration intercepts requests before they reach the provider and classifies responses through the nine-guard G-series:

Provider	Class	BAA status
Anthropic	AnthropicGovernance	Available (enterprise)
OpenAI	OpenAIGovernance	Available (enterprise only)
Google Gemini	GeminiGovernance	Not available (consumer tier)
Azure OpenAI	AzureOpenAIGovernance	In scope (Microsoft BAA)
AWS Bedrock	BedrockGovernance	In scope (AWS BAA)
Ollama (local)	OllamaGovernance	Not required (self-hosted)

from connexum_governance import GovernanceClient
from connexum_governance.integrations.anthropic import AnthropicGovernance

gov = GovernanceClient(api_url="http://localhost:3100", license_key="your-key")
ag = AnthropicGovernance(client=gov, agent_name="my-agent")

# Check before sending
decision = ag.intercept_request({
    "model": "claude-3-haiku-20240307",
    "max_tokens": 512,
    "messages": [{"role": "user", "content": "Hello"}],
})

# Or use the proxy to intercept transparently
import anthropic
governed = ag.wrap(anthropic.Anthropic())
response = governed.messages.create(
    model="claude-3-haiku-20240307",
    max_tokens=512,
    messages=[{"role": "user", "content": "Hello"}],
)

The same pattern applies to all six providers. See each integration module in connexum_governance/integrations/ for provider-specific usage and BAA configuration details.

Orchestrator Integrations

Optional wrappers for popular AI orchestrators. Install the extras for the frameworks you use:

pip install connexum-governance[langgraph]
pip install connexum-governance[crewai]
pip install connexum-governance[autogen]
pip install connexum-governance[langchain]
pip install connexum-governance[langgraph]
pip install connexum-governance[autogen]
pip install connexum-governance[claude-agent-sdk]
pip install connexum-governance[google-adk]
pip install connexum-governance[bedrock-agentcore]
pip install connexum-governance[haystack]
pip install connexum-governance[llamaindex]

All nine orchestrator integrations cover six canonical governance hook surfaces: agentStart, toolCall, toolResult, llmCall, llmResult, agentEnd.

Orchestrator	Class	Framework
LangGraph	GovernanceTool	langgraph
CrewAI	GovernanceCallback	crewai
AutoGen	GovernanceGuardrail	pyautogen
LangChain	GovernanceCallbackHandler	langchain
Claude Agent SDK	ClaudeAgentGovernance	claude-agent-sdk
Google ADK	GoogleAdkGovernance	google-adk
Bedrock AgentCore	BedrockAgentCoreGovernance	boto3 / bedrock-agentcore
Haystack (deepset)	HaystackGovernance	haystack-ai
LlamaIndex	LlamaIndexGovernance	llama-index

Framework-specific highlights

LangChain framework adapter (create_governed_langchain, GovernedTool): Full framework adapter mirroring the TypeScript governed-langchain.ts architecture. Two enforcement layers for defense-in-depth:

1. GovernedTool / governed_tool() -- wraps any LangChain tool's _run() and _arun() methods. This is the GUARANTEED BLOCKING PRIMITIVE: governance runs inside tool invocation regardless of executor type or LangChain version.

2. GovernanceCallbackHandler / AsyncGovernanceCallbackHandler -- observability hooks wired via callbacks=[handler]. Best-effort blocking; LangChain may swallow exceptions in some executor configurations (version-dependent).

import os
from connexum_governance.integrations.langchain_integration import (
    create_governed_langchain,
)

gov = create_governed_langchain(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1 (GUARANTEED BLOCKING): wrap individual tools
safe_search_tool = gov.governed_tool(raw_search_tool)
executor = AgentExecutor(agent=agent, tools=[safe_search_tool])

# Option 2 (defense-in-depth): wrap the entire executor's callbacks
executor = gov.wrap_agent(executor)

# Option 3 (LangGraph): wrap a StateGraph to intercept ToolNode calls
graph = gov.wrap_graph(state_graph)

CRITICAL ARCHITECTURAL NOTE: LangChain's callback system is OBSERVABILITY-FIRST, not BLOCKING-FIRST. The TRUE blocking primitive is GovernedTool._run() / _arun(), not the callback handler. Use governed_tool() for tools you need blocked; use the callback handler for audit coverage. See module docstring in connexum_governance/integrations/langchain_integration.py for full details.

Action name namespace for pack rule targeting:

• framework.langchain.agent_action -- on_agent_action
• framework.langchain.tool_invoke -- on_tool_start + GovernedTool._run()
• framework.langchain.llm_start -- on_llm_start (includes injection scan)
• framework.langchain.graph.tool_node -- wrap_state_graph ToolNode

Optional dependencies (install to use with real LangChain):

pip install connexum-governance[langchain]  # includes langgraph >= 0.0.30

Version compat range tested: langchain >= 0.2.0, < 0.4.0; langchain-core >= 0.2.0, < 0.4.0; langgraph >= 0.0.30, < 0.4.0.

OpenAI Agents SDK framework adapter (create_governed_openai_agents, Sprint 5, WS-08):

The OpenAI Agents SDK (Python, released March 2025) provides FIRST-CLASS blocking primitives via input_guardrails and output_guardrails. This is a materially stronger blocking story than LangChain's callback system.

Framework	Primary blocking mechanism	Blocking guarantee
LangChain	GovernedTool._run() (tool wrapping required)	Version-dependent for callbacks; guaranteed for GovernedTool
OpenAI Agents SDK	input_guardrails / output_guardrails (SDK contract)	FIRST-CLASS: SDK raises named exception on tripwire

The SDK's guardrail contract: when a guardrail returns tripwire_triggered=True, the SDK raises InputGuardrailTripwireTriggered or OutputGuardrailTripwireTriggered and the run is stopped. This is part of the SDK's public API contract -- not a side-effect of exception propagation, not version-dependent.

Three enforcement layers for defense-in-depth:

1. GovernanceInputGuardrail -- PRIMARY BLOCKING. Attaches to input_guardrails. Runs governance check + prompt injection scan before the agent processes input. Tripwire blocks the entire run.

2. GovernanceOutputGuardrail -- PRIMARY BLOCKING. Attaches to output_guardrails. Scans output for PII / PHI / credentials (G2/G3 equivalent) and runs governance check. Tripwire blocks the output from being returned.

3. GovernanceAgentHooks -- OBSERVABILITY + DEFENSE-IN-DEPTH. Attaches to agent hooks. on_tool_start raises GovernanceViolation on DENY (defense-in-depth alongside guardrails). on_handoff logs handoff boundary events to audit chain. on_tool_end scans tool output for sensitive data.

4. governed_function_tool() decorator -- DEFINITION-TIME BLOCKING. Wraps tool functions at definition time so governance runs before execution regardless of which agent or runner the tool is registered on. Mirrors GovernedTool._run() in the LangChain adapter.

import os
from connexum_governance.integrations.openai_agents_integration import (
    create_governed_openai_agents,
    governed_function_tool,
)

gov = create_governed_openai_agents(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1: attach when defining an agent
from openai_agents import Agent
agent = Agent(
    name="my-agent",
    instructions="You are a helpful assistant.",
    input_guardrails=[gov.input_guardrail],   # primary blocking
    output_guardrails=[gov.output_guardrail], # primary blocking + PII scan
    hooks=gov.hooks,                          # audit trail + defense-in-depth
)

# Option 2: wrap an existing agent (idempotent, safe to call multiple times)
agent = gov.wrap_agent(existing_agent)

# Option 3: wrap tools at definition time (guaranteed blocking at tool boundary)
tool_decorator = gov.function_tool_decorator()

@tool_decorator
async def web_search(query: str) -> str:
    return f"Search results for: {query}"

# Inject the tool into your agent
agent = Agent(name="searcher", tools=[web_search])

Action name namespace for pack rule targeting:

• framework.openai_agents.input_guardrail -- GovernanceInputGuardrail
• framework.openai_agents.output_guardrail -- GovernanceOutputGuardrail
• framework.openai_agents.tool_invoke -- on_tool_start hooks + governed_function_tool
• framework.openai_agents.handoff -- on_handoff hook (audit logging)

Optional dependencies:

pip install connexum-governance[openai-agents]  # openai-agents >= 0.0.3

Version compat range tested: openai-agents >= 0.0.3, < 2.0.

LlamaIndex framework adapter (create_governed_llama_index, Sprint 5, WS-09):

LlamaIndex is Python's primary RAG and document-retrieval orchestration framework. This adapter intercepts at three layers for defense-in-depth:

Layer	Mechanism	Blocking guarantee
Tool dispatch	GovernedFunctionTool.__call__ / acall	GUARANTEED: runs inside tool dispatch path
Query entry	wrap_query_engine()	GUARANTEED: blocks before any retrieval
Chat entry	wrap_chat_engine()	GUARANTEED: blocks before any retrieval
LLM events	GovernanceLlamaIndexCallback	BEST-EFFORT: observability (version-dependent)

The GUARANTEED BLOCKING PRIMITIVE is GovernedFunctionTool. The callback handler (GovernanceLlamaIndexCallback) fires for LLM calls, agent steps, and retrievals but may be swallowed by LlamaIndex's try/except wrappers in some versions. Use the callback for audit coverage; use GovernedFunctionTool for guaranteed enforcement.

import os
from connexum_governance.integrations.llama_index_integration import (
    create_governed_llama_index,
)

gov = create_governed_llama_index(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1 (GUARANTEED BLOCKING): wrap individual tools for agents
safe_tool = gov.governed_tool(my_function_tool)
agent = ReActAgent.from_tools([safe_tool], llm=llm)

# Option 2 (query-level gate): wrap the query engine
governed_engine = gov.wrap_query_engine(index.as_query_engine())
response = governed_engine.query("What are the HIPAA retention requirements?")
# GovernanceViolation raised on DENY before any retrieval runs

# Option 3 (chat-level gate): wrap the chat engine
governed_chat = gov.wrap_chat_engine(index.as_chat_engine())
response = governed_chat.chat("Summarize the data retention policy.")

# Option 4 (defense-in-depth): register callback for audit coverage
from llama_index.core.callbacks import CallbackManager
callback_manager = CallbackManager([gov.callback])

Action name namespace for pack rule targeting:

• framework.llamaindex.tool_call -- GovernedFunctionTool dispatch
• framework.llamaindex.query -- wrap_query_engine entry + RETRIEVE events
• framework.llamaindex.chat_message -- wrap_chat_engine entry
• framework.llamaindex.agent_step -- callback on_event_start (AGENT_STEP, LLM, etc.)

Optional dependencies:

pip install connexum-governance[llama-index]
# installs: llama-index-core >= 0.10.0, llama-index >= 0.10.0

Version compat range tested: llama-index-core >= 0.10.0, llama-index >= 0.10.0.

CrewAI framework adapter (create_governed_crewai, Sprint 5, WS-09):

CrewAI is a role-based multi-agent framework built on top of LangChain. This adapter intercepts at tool, agent, and crew levels:

Layer	Mechanism	Blocking guarantee
Tool dispatch	GovernanceCrewAITool._run()	GUARANTEED: runs before tool execution
Crew kickoff	wrap_crew() kickoff wrap	GUARANTEED: blocks before crew starts
Agent steps	step_callback	OBSERVABILITY: fires AFTER step (cannot block initiating call)
Task outputs	task_callback	OBSERVABILITY: fires AFTER task completion

CRITICAL ARCHITECTURAL NOTE: CrewAI's step_callback and task_callback are OBSERVABILITY hooks that fire AFTER execution. They can halt the crew run when raise_on_deny=True is set, but they cannot block the action that already ran. The GUARANTEED BLOCKING PRIMITIVE is GovernanceCrewAITool._run().

import os
from connexum_governance.integrations.crewai_integration import (
    create_governed_crewai,
)

gov = create_governed_crewai(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1 (GUARANTEED BLOCKING): wrap individual tools
safe_tool = gov.governed_tool(raw_search_tool)
agent = Agent(role="researcher", tools=[safe_tool], llm=llm, ...)

# Option 2 (agent-level): wrap all tools on an agent at once
agent = gov.wrap_agent(agent)

# Option 3 (crew-level, full defense-in-depth):
# Wraps all tools on all agents + injects step/task callbacks + wraps kickoff
crew = gov.wrap_crew(crew)
result = crew.kickoff()
# GovernanceViolation raised if crew_kickoff is denied before any agent runs

# Option 4 (manual callback injection):
step_cb = gov.make_step_callback(raise_on_deny=False)  # observe-only
task_cb = gov.make_task_callback(raise_on_deny=True)   # halt on violation
crew = Crew(agents=[...], tasks=[...], step_callback=step_cb, task_callback=task_cb)

Action name namespace for pack rule targeting:

• framework.crewai.tool_run -- GovernanceCrewAITool._run() (GUARANTEED)
• framework.crewai.crew_kickoff -- wrap_crew kickoff gate (GUARANTEED)
• framework.crewai.agent_step -- step_callback (OBSERVABILITY)
• framework.crewai.task_complete -- task_callback (OBSERVABILITY)

Optional dependencies:

pip install connexum-governance[crewai]
# installs: crewai >= 0.30.0

Version compat range tested: crewai >= 0.30.0.

AutoGen v0.4 framework adapter (create_governed_autogen, Sprint 5, WS-10):

AutoGen v0.4 (autogen-agentchat / autogen-core) is async-first and uses message passing between agents. This adapter targets the REWRITTEN v0.4 API only. The deprecated pyautogen package (v0.2.x, synchronous) is NOT supported.

IMPORTANT: AutoGen v0.4 (autogen-agentchat >= 0.4.0) is a ground-up rewrite with a new API surface. If you are importing from pyautogen, migrate to the new packages before using this adapter.

Layer	Mechanism	Blocking guarantee
Tool dispatch	GovernedAutoGenTool.run() / run_json()	GUARANTEED: runs inside tool dispatch path
Agent message boundary	wrap_chat_agent() on_messages intercept	GUARANTEED: blocks before agent processes message
Conversation-level	wrap_group_chat() run/run_stream wrap	BEST-EFFORT per conversation turn
Graceful cancel	CancellationToken.cancel() on DENY	COOPERATIVE: AutoGen-native cancellation

The GUARANTEED BLOCKING PRIMITIVE is GovernedAutoGenTool.run() / run_json(). CancellationToken integration provides cooperative AutoGen-native cancellation on top of GovernanceViolation raises (configurable via cancel_on_deny=True).

import os
from connexum_governance.integrations.autogen_integration import (
    create_governed_autogen,
    governed_autogen_tool,
)

gov = create_governed_autogen(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1 (GUARANTEED BLOCKING): wrap individual tools
from autogen_core.tools import FunctionTool

async def search_web(query: str) -> str:
    return f"Results for: {query}"

raw_tool = FunctionTool(search_web, description="Search the web")
safe_tool = gov.governed_tool(raw_tool)

# Option 2 (agent message boundary): wrap the agent
from autogen_agentchat.agents import AssistantAgent
agent = AssistantAgent("assistant", tools=[safe_tool], model_client=client)
gov.wrap_agent(agent)  # wraps on_messages + auto-wraps agent tools

# Option 3 (conversation-level): wrap the group chat
from autogen_agentchat.teams import RoundRobinGroupChat
team = gov.wrap_group_chat(RoundRobinGroupChat([agent1, agent2]))
await team.run(task="Analyze this patient data")

# Option 4 (graceful cancellation on DENY):
gov_cancel = create_governed_autogen(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    cancel_on_deny=True,  # calls CancellationToken.cancel() before raising
)

Action name namespace for pack rule targeting:

• framework.autogen.tool_run -- GovernedAutoGenTool.run() / run_json() (GUARANTEED)
• framework.autogen.message_received -- wrap_chat_agent on_messages (GUARANTEED)
• framework.autogen.group_chat_run -- wrap_group_chat pre-run check (BEST-EFFORT)
• framework.autogen.handoff -- handoff events detected in streaming runs

Optional dependencies:

pip install connexum-governance[autogen]
# installs: autogen-agentchat >= 0.4.0, autogen-core >= 0.4.0
# Do NOT install pyautogen alongside autogen-agentchat (conflict)

Version compat range tested: autogen-agentchat >= 0.4.0; autogen-core >= 0.4.0.

LangGraph standalone adapter (create_governed_langgraph, Sprint 5, WS-10):

LangGraph is Python's primary stateful agent graph framework. This dedicated adapter provides richer governance than the lightweight wrap_state_graph() in the LangChain adapter and is the recommended choice for LangGraph-first customers.

The LangChain adapter's wrap_state_graph() is now a thin re-export of the canonical implementation in this module. Existing customers are not broken.

Layer	Mechanism	Blocking guarantee
Per-node	GovernedStateGraph.add_node() wraps each node	GUARANTEED: fires before every node execution
Per-invocation	GovernedStateGraph.compile() -> governed Pregel	GUARANTEED: one check per graph.invoke() call
Edge routing	GovernedStateGraph.add_conditional_edges() wraps routing fn	BEST-EFFORT routing check before transition
Post-compile	wrap_pregel() on already-compiled Pregel	GUARANTEED: same as compile-level gate

New additions vs the LangChain adapter's wrap_state_graph():

• compile()-level Pregel invoke() / ainvoke() governance (new)
• Conditional edge routing function governance (new)
• GovernedStateGraph proxy class for clean graph construction (new)
• wrap_pregel() for post-compile interception (new)

import os
from connexum_governance.integrations.langgraph_integration import (
    create_governed_langgraph,
    GovernedStateGraph,
)

gov = create_governed_langgraph(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

# Option 1 (recommended for new graphs): GovernedStateGraph proxy
from langgraph.graph import StateGraph
raw_graph = StateGraph(MyState)
g = gov.governed_state_graph(raw_graph)
g.add_node("agent", agent_fn)      # wrapped with per-node governance
g.add_node("tools", tool_node)     # uses 'framework.langgraph.tool_node' action
g.add_conditional_edges("agent", routing_fn)  # routing function governed
g.add_edge("tools", "agent")
app = g.compile()                  # returns governed Pregel
result = app.invoke({"messages": [...]})  # invoke-level governance check

# Option 2 (existing graph): wrap_graph() in place
raw_graph = StateGraph(MyState)
gov.wrap_graph(raw_graph)
app = raw_graph.compile()  # compile() now returns governed Pregel

# Option 3 (post-compile): wrap an already-compiled Pregel
app = raw_graph.compile()
app = gov.wrap_pregel(app)

# Backwards compatibility: wrap_state_graph from langchain_integration still works
from connexum_governance.integrations.langchain_integration import wrap_state_graph
wrap_state_graph(raw_graph, governance_server_url="...", license_key="...")

Action name namespace for pack rule targeting:

• framework.langgraph.invoke -- compile-level invoke/ainvoke gate (GUARANTEED)
• framework.langgraph.node_execute -- per-node governance (GUARANTEED)
• framework.langgraph.tool_node -- ToolNode-specific node check (GUARANTEED)
• framework.langgraph.edge_transition -- conditional edge routing (BEST-EFFORT)

NOTE: The old action framework.langchain.graph.tool_node (from the LangChain adapter's inline wrap_state_graph) is replaced by framework.langgraph.tool_node. Update pack rules targeting the old action name.

Optional dependencies:

pip install connexum-governance[langgraph]
# installs: langgraph >= 0.0.30

Version compat range tested: langgraph >= 0.0.30, < 0.4.0.

Pydantic AI framework adapter (create_governed_pydantic_ai, Sprint 5, WS-13):

Pydantic AI is a type-safe agent SDK with no built-in callback or hook system. This adapter works entirely through function wrapping:

Layer	Mechanism	Blocking guarantee
Tool dispatch	governed_tool_decorator() / governed_tool_plain_decorator()	GUARANTEED: governance baked into tool function at registration time
Agent entry	wrap_agent_run() -- wraps run/run_sync/run_stream	GUARANTEED: blocks before any LLM call; includes injection scan
Output side	GovernedAgentResult.validate()	OBSERVABILITY: scans result for PII/PHI after agent produces output

CRITICAL ARCHITECTURAL NOTE: Pydantic AI has NO callback system, no guardrail protocol, no hook interface. ALL enforcement must be done through function wrapping. governed_tool_decorator() replaces @agent.tool so tools carry governance from definition time. Call wrap_agent(agent) BEFORE registering tools.

import os
from connexum_governance.integrations.pydantic_ai_integration import (
    create_governed_pydantic_ai,
)
from pydantic_ai import Agent

gov = create_governed_pydantic_ai(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

agent = Agent("openai:gpt-4o", system_prompt="You are a healthcare assistant.")

# RECOMMENDED: wrap agent first, THEN register tools
gov.wrap_agent(agent)  # wraps run/run_sync/run_stream + replaces tool decorators

@agent.tool              # now governed (decorator was replaced by wrap_agent)
async def lookup_patient(ctx, mrn: str) -> dict:
    ...

result = await agent.run("Summarize patient MRN 12345")
safe_result = gov.validate_result(result)  # output-side PII/PHI scan

Action name namespace for pack rule targeting:

• framework.pydantic_ai.tool_invoke -- governed tool execution (GUARANTEED)
• framework.pydantic_ai.agent_run -- run/run_sync entry guard (GUARANTEED)
• framework.pydantic_ai.result_validate -- GovernedAgentResult output scan
• framework.pydantic_ai.stream_chunk -- run_stream per-chunk audit

CANONICAL TOOL-NAME WIRE FORM (IMPORTANT FOR PACK AUTHORS): Pydantic AI tools are sent to gov-server with toolName='tool.<name>' to preserve the framework's tool-namespace convention. Pack rules targeting bare tool names will not match; use prefix matching or include the tool. prefix in your rule. The adapter also sends action='framework.pydantic_ai.tool_invoke' in the same payload, so rules may target either surface.

Optional dependencies:

pip install connexum-governance[pydantic-ai]  # pydantic-ai >= 0.0.13

Version compat range tested: pydantic-ai >= 0.0.13.

Haystack v2 framework adapter (create_governed_haystack, Sprint 5, WS-13):

Haystack v2 (haystack-ai) is Python's primary RAG and clinical document Q&A framework. IMPORTANT: This adapter targets haystack-ai >= 2.0.0 ONLY. The deprecated Farm-Haystack v1.x package is NOT supported.

Haystack v2 is pipeline-first. Governance applies at four layers:

Layer	Mechanism	Blocking guarantee
Pipeline entry	wrap_pipeline() -- wraps Pipeline.run()	GUARANTEED: one check before any component runs
Per-component	GovernedComponent -- wraps any component's run()	GUARANTEED: fires before each component execution
Tool dispatch	wrap_tool_invoker() -- wraps ToolInvoker.run()	GUARANTEED: per-tool-call gate in agentic pipelines
LLM boundary	wrap_generator() -- wraps any Generator run()	DEFENSE-IN-DEPTH: injection scan + check before LLM call

HEALTHCARE NOTE: For HIPAA-compliant clinical document Q&A pipelines, a single wrap_pipeline() call covers all clinical RAG query traffic through the pipeline via the framework.haystack.pipeline_run action namespace.

import os
from connexum_governance.integrations.haystack_integration import (
    create_governed_haystack,
)
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.retrievers import InMemoryBM25Retriever

gov = create_governed_haystack(
    governance_server_url="http://localhost:3200",
    license_key=os.environ["MYCC_LICENSE_KEY"],
    pack_ids=["hipaa"],
)

pipeline = Pipeline()

# Option 1 (RECOMMENDED for clinical RAG): pipeline-level gate
gov.wrap_pipeline(pipeline)  # one check per pipeline.run() call

# Option 2 (defense-in-depth): govern individual components
retriever = InMemoryBM25Retriever(document_store=store)
pipeline.add_component("retriever", gov.governed_component(retriever))

# Option 3 (agentic pipelines with tool calls): wrap ToolInvoker
from haystack.components.routers import ToolInvoker
invoker = ToolInvoker(tools=[...])
gov.wrap_tool_invoker(invoker)

# Option 4 (LLM boundary injection scan): wrap the generator
llm = OpenAIGenerator(model="gpt-4o")
gov.wrap_generator(llm)  # injection scan + governance check before LLM call

Action name namespace for pack rule targeting:

• framework.haystack.pipeline_run -- Pipeline.run() gate (GUARANTEED)
• framework.haystack.component_run -- GovernedComponent.run() (GUARANTEED)
• framework.haystack.tool_invoke -- wrap_tool_invoker (GUARANTEED)
• framework.haystack.generator_call -- wrap_generator (DEFENSE-IN-DEPTH)

Optional dependencies:

pip install connexum-governance[haystack]  # haystack-ai >= 2.0.0

Version compat range tested: haystack-ai >= 2.0.0. Farm-Haystack v1.x NOT supported.

9-framework blocking story summary:

Framework	Primary blocking primitive	Blocking contract	Notes
OpenAI Agents SDK	input_guardrails / output_guardrails	SDK public API contract	Strongest: tripwire is first-class SDK feature
AutoGen v0.4	GovernedAutoGenTool.run() / run_json()	Guaranteed inside tool dispatch	+ CancellationToken cooperative cancel
Pydantic AI	governed_tool_decorator() + wrap_agent_run()	Guaranteed at tool definition + agent entry	No hook system; all enforcement via function wrapping
Haystack v2	wrap_pipeline() + GovernedComponent.run()	Guaranteed (pipeline gate + per-component)	Healthcare RAG; framework.haystack.pipeline_run = HIPAA gate
LangChain	GovernedTool._run()	Guaranteed inside tool path	Callbacks are observability-only
LlamaIndex	GovernedFunctionTool.__call__	Guaranteed inside tool path	Same pattern as LangChain GovernedTool
CrewAI	GovernanceCrewAITool._run()	Guaranteed inside tool path	Callbacks fire post-execution
LangGraph	GovernedStateGraph compile-level + per-node	Guaranteed (both layers)	Dedicated adapter; richer than LangChain's wrap_state_graph
LangGraph (via LC)	wrap_state_graph (re-export)	Guaranteed per-node + compile	Backwards compat; uses LangGraph adapter under the hood

LangChain (GovernanceCallbackHandler in integrations/langchain.py): Original callback-based integration. Subclasses LangChain's callback interface. Wire into any chain via callbacks=[handler]. Covers on_chain_start, on_tool_start, on_tool_end, on_llm_start, on_llm_end, on_chain_end.

Claude Agent SDK (ClaudeAgentGovernance): Pack-binding primacy enforced at agent spawn. Phantom-agent detection (DA-MED-NEW-1). Runaway counter. G5/G7 exfiltration scan on tool arguments.

Google ADK (GoogleAdkGovernance): Vertex AI BAA boundary enforcement. Direct AI Studio calls denied when require_vertex_ai=True under HIPAA pack. Session context (session_id, user_id) forwarded in all audit events.

Bedrock AgentCore (BedrockAgentCoreGovernance): HIPAA BAA region enforcement. Approved regions: us-east-1, us-west-2, eu-west-1, ap-southeast-2. Action groups (Bedrock's tool concept) map to toolCall/toolResult hooks.

Haystack (HaystackGovernance): Pipeline-based architecture. Eval-mode pipelines write to a separate eval audit path. Multi-backend PromptNode: each backend invocation is a distinct canonical event. GDPR default pack.

LlamaIndex (LlamaIndexGovernance): Document retrieval is a first-class ingress surface. Healthcare document classes (MedicalRecord, LabReport, etc.) trigger HIPAA compliance checks. ReAct trace forwarded in audit events. Response synthesis classified for distilled PHI.

See connexum_governance/integrations/ for full usage examples and per-framework BAA configuration details.

Running Tests

Unit tests (mock transport, no server required)

cd packages/python-sdk
pytest

Integration tests (real TypeScript governance server)

Integration tests start the TypeScript governance server as a subprocess and run check_action calls against it. The server build must exist in packages/governance-server/src/index.ts and npx tsx must be available.

# From the repo root
cd packages/python-sdk
pytest -m integration

The fixture in tests/integration/conftest.py handles server start/stop automatically. Port 3200 must be free. The server runs with DEFAULT_DENY=true and HITL_ENABLED=true (no API_TOKEN required).

To run all tests including integration:

pytest -m "integration or not integration"
# or simply:
pytest

Integration tests are excluded from the default run (pytest without -m) only if your pytest.ini or pyproject.toml marks them as such. They are safe to run at any time as long as the governance server source is present.

---

Adapter Contract Test Suite

Overview

tests/contract/ applies a uniform set of invariants to every adapter registered in the Python SDK. Each adapter registers by contributing to the ADAPTER_CONFIGS or FRAMEWORK_CONFIGS list. Running the suite proves that an adapter correctly integrates with the governance runtime.

LLM Provider Contract (tests/contract/test_llm_provider_contract.py)

TestLLMProviderContract is a @pytest.mark.parametrize class applying 13 invariants to each AdapterConfig registered in ADAPTER_CONFIGS.

Invariants:

#	Name	What it checks
1	constructor_valid	Client builds without error
2	api_url_stored	api_url field persists correctly
3	license_key_stored	license_key field persists correctly
4	allow_no_violation	ALLOW: check_action completes without exception
5	deny_raises_governance_violation	DENY + enforce: GovernanceViolation raised
6	pending_raises_pending_approval	PENDING + enforce: GovernancePendingApproval raised
7	governance_check_fires	At least one HTTP request to check-action or check-task captured
8	action_name_in_payload	tool, toolName, or action key matches documented action name
9	fail_open_unreachable	Server unreachable + fail-open: call proceeds without exception
10	fail_closed_unreachable	Server unreachable + fail-closed: GovernanceViolation raised
11	license_key_in_auth_header	Authorization: Bearer <key> present on governance requests
12	enforce_false_no_raise	DENY with enforce=False: no exception raised
13	allow_with_tool_payload	Call with tool payload: check fires with tool data

Adapter registered (Anthropic):

AdapterConfig(
    name="Anthropic",
    adapter_factory=lambda transport: AnthropicGovernance(
        client=make_client(transport),
        _check_fn=None,
    ),
    minimal_request={"model": "claude-3-haiku-20240307", "max_tokens": 1, "messages": [...]},
    request_with_tool={"tools": [{"name": "test_tool", ...}], ...},
    expected_tool_name="test_tool",
    expected_action_name="llm.chat",
    raises_on_enforce=False,  # Returns GovernanceDecision, does not raise
)

How to register a new LLM adapter:

Add an entry to ADAPTER_CONFIGS in test_llm_provider_contract.py:

ADAPTER_CONFIGS.append(AdapterConfig(
    name="MyProvider",
    adapter_factory=lambda transport: MyProviderGovernance(
        client=make_client(transport),
    ),
    minimal_request={"model": "my-model", "prompt": "test"},
    request_with_tool={"tools": [{"name": "my_tool"}], ...},
    expected_tool_name="my_tool",
    expected_action_name="llm.myprovider.call",
    raises_on_enforce=True,
))

---

Framework Adapter Contract (tests/contract/test_framework_adapter_contract.py)

TestFrameworkAdapterContract applies 9 invariants to each FrameworkAdapterConfig registered in FRAMEWORK_CONFIGS.

Invariants:

#	Name	What it checks
1	constructor_valid	Adapter builds without error
2	allow_no_violation	ALLOW: invoke_tool completes without GovernanceViolation
3	deny_blocks_call	DENY: GovernanceViolation raised or denial string returned
4	pending_no_crash	PENDING: completes or raises approved exception type
5	governance_check_fires	At least one check_action or check_task request captured
6	action_namespace	Action key matches documented adapter action name/prefix
7	constructor_validates_required_params	Passing None/empty rejects at construction or first use
8	async_sync_consistent	Async variant behaves consistently with sync on ALLOW
9	public_factory_type	Factory returns a non-None, callable-backed adapter instance

Framework adapters registered (all 8):

Adapter	Primary path	Action key sent
LangChain	GovernanceCallbackHandler.on_tool_start	tool.<name>
OpenAIAgents	OpenAIAgentsGovernance.agent_start + tool_call	tool.<name>
LlamaIndex	LlamaIndexGovernance.tool_call(session_id, name, args)	tool.<name>
CrewAI	GovernanceCallback.on_tool_start	<name> (raw, DIVERGENCE-005)
AutoGen	GovernanceGuardrail.protect(name) decorator	<name> (raw, DIVERGENCE-005)
LangGraph	GovernanceTool.invoke	<wrapped_tool.name> (raw, DIVERGENCE-005)
PydanticAI	governed_tool_decorator (DIVERGENCE-002)	framework.pydantic_ai.tool_invoke
HaystackV2	HaystackGovernance.pipeline_start + node_start	agent.start / node.<type>

Divergences documented:

ID	Adapter	Description	Status
DIVERGENCE-002	PydanticAI	Now accepts GovernanceClient instance via governance_client= (preferred). Standalone governance_server_url / license_key params still work but emit DeprecationWarning.	RESOLVED (WS-05 ITEM 2)
DIVERGENCE-003	LangChain	on_tool_start now returns GovernanceDecision on ALLOW. LangChain ignores the return value internally; this is for callers invoking the handler directly.	RESOLVED (WS-05 ITEM 3)
DIVERGENCE-004	LangChain	on_tool_start DENY does NOT block tool execution even with raise_on_deny=True. This is an architectural reality of LangChain's callback system (not a bug). GovernedTool._run() is the GUARANTEED blocking primitive. See module docstring.	INTENTIONAL (WS-05 ITEM 4)
DIVERGENCE-005	CrewAI, AutoGen, LangGraph	Now send tool.<name> (with tool. prefix), matching OpenAIAgents / LlamaIndex / LangChain.	RESOLVED (WS-05 ITEM 5)
DIVERGENCE-006	OpenAIAgents	Now auto-initializes default agent state on first tool_call() if agent_start() was not called. A warning is written to stderr. Explicit agent_start() is still recommended for strict lifecycle enforcement.	RESOLVED (WS-05 ITEM 6)
DIVERGENCE-007	PydanticAI	Now sends both tool key (tool.<name>) AND action key (framework.pydantic_ai.tool_invoke). Both are in the payload so pack rules can target either.	RESOLVED (WS-05 ITEM 7)

---

Running the contract suite

# All contract tests
python -m pytest tests/contract/ -v

# LLM provider suite only
python -m pytest tests/contract/test_llm_provider_contract.py -v

# Framework suite only
python -m pytest tests/contract/test_framework_adapter_contract.py -v

# Single adapter
python -m pytest tests/contract/test_framework_adapter_contract.py -k "LangGraph" -v