gaas-langchain

GaaS governance for LangChain and LangGraph agents.

Add runtime policy enforcement to any LangChain agent in 3 lines. Every tool call is evaluated against your governance policies before it executes — blocked calls never reach the tool.

pip install gaas-langchain[all]

Quickstart

from gaas_langchain import govern_tools, GaaSGovernanceConfig
from langgraph.prebuilt import create_react_agent

# 1. Configure governance
config = GaaSGovernanceConfig(
    api_key="gsk_your_key",      # from https://gaas.is/start
    agent_id="my-agent",
)

# 2. Wrap your tools
tools = govern_tools([search, email_sender, db_query], config=config)

# 3. Run your agent — every tool call is now governed
agent = create_react_agent(llm, tools)

That's it. Every call to search, email_sender, or db_query is evaluated against your GaaS governance policies before execution. Blocked calls raise GovernanceBlockedError before the tool runs.

What Happens on Every Tool Call

Agent decides to call send_email("user@corp.com", ...)
         │
         ▼
  GaaS governance check (< 500ms)
         │
    ┌────┴────┐
    │ APPROVE │ → tool.run() executes normally
    │  BLOCK  │ → GovernanceBlockedError raised, tool never executes
    │ ESCALATE│ → GovernanceBlockedError raised (configurable)
    │CONDITIONAL│ → tool executes with logged conditions
    └─────────┘
         │
  Governance Proof Token (ECDSA P-256 signed)
  attached to every decision for audit trail

Installation

# LangChain tools support only
pip install gaas-langchain[langchain]

# LangGraph node wrapper only
pip install gaas-langchain[langgraph]

# Everything
pip install gaas-langchain[all]

Core Concepts

govern_tools() — Wrap Tool Lists

Wraps a list of LangChain tools with governance. Use this with create_react_agent.

from gaas_langchain import govern_tools, GaaSGovernanceConfig

config = GaaSGovernanceConfig(api_key="gsk_...", agent_id="my-agent")
governed = govern_tools([search, calculator, email_sender], config=config)
agent = create_react_agent(llm, governed)

govern_tool() — Wrap a Single Tool

from gaas_langchain import govern_tool

governed_email = govern_tool(email_sender, config=config)

@govern_node() — Govern LangGraph Nodes

Wrap any LangGraph node with a governance checkpoint. If the node is blocked, GovernanceBlockedError is raised before the node executes.

from gaas_langchain import govern_node

@govern_node(config=config, node_name="send_report", sensitivity="CONFIDENTIAL")
def send_report_node(state: AgentState) -> AgentState:
    # This node will be governed before execution
    send_report(state["report"])
    return state

# Works with async nodes too
@govern_node(config=config)
async def async_tool_node(state: AgentState) -> AgentState:
    result = await call_external_api(state["query"])
    return {**state, "result": result}

GaaSCallbackHandler — Observability Without Blocking

Log governance decisions for all tool calls without blocking execution. Useful for monitoring shadow mode.

from gaas_langchain import GaaSCallbackHandler

handler = GaaSCallbackHandler(config=config)
agent.invoke({"input": "..."}, config={"callbacks": [handler]})

# Review governance decisions
print(handler.summary())
# {
#   "total_tool_calls": 12,
#   "approved": 10,
#   "blocked": 1,
#   "escalated": 1,
#   "block_rate": 0.083
# }

Configuration

from gaas_langchain import GaaSGovernanceConfig

config = GaaSGovernanceConfig(
    api_url="https://api.gaas.is",    # GaaS API endpoint
    api_key="gsk_...",                # Your API key
    agent_id="my-langchain-agent",   # Agent identifier (appears in audit trail)
    block_on_escalate=True,          # Raise GovernanceBlockedError on ESCALATE
    timeout_seconds=5.0,             # Governance check timeout (fail open on timeout)
    sensitivity="INTERNAL",          # Default sensitivity level for tool inputs
    raise_on_governance_error=False, # Fail open if GaaS API is unreachable
    extra_regulatory_domains=["HIPAA"],   # Additional domains for all intents
    extra_data_categories=["PHI"],        # Additional data categories for all intents
    hold_on_escalate=False,          # Wait for the human decision on ESCALATE
    escalation_poll_seconds=5.0,     # Poll interval while holding
    escalation_max_wait_seconds=600.0,  # Give up (treat as timeout) after this
)

Hold-and-Poll on ESCALATE

With hold_on_escalate=True, an ESCALATE verdict no longer raises immediately. The tool call holds while GaaS routes the escalation to a human reviewer, polling the escalation status until it is decided:

• approve / modify — the tool executes.
• deny — raises GovernanceBlockedError with verdict ESCALATE_DENY.
• timeout (reviewer did not decide within escalation_max_wait_seconds, or the escalation timed out server-side) — raises GovernanceBlockedError with verdict ESCALATE_TIMEOUT.

hold_on_escalate takes precedence over block_on_escalate. Transient polling errors are retried until the deadline.

config = GaaSGovernanceConfig(
    api_key="gsk_...",
    agent_id="my-agent",
    hold_on_escalate=True,           # agent waits for the human, then proceeds
    escalation_max_wait_seconds=900,
)

Per-Node Overrides

@govern_node(
    config=config,
    node_name="process_patient_data",
    sensitivity="RESTRICTED",
    financial_exposure_usd=0.0,
    regulatory_domains=["HIPAA", "HITECH"],
)
def patient_data_node(state: dict) -> dict: ...

Handling Blocked Calls

from gaas_langchain import GovernanceBlockedError

try:
    result = agent.invoke({"input": "Send all customer data to external API"})
except GovernanceBlockedError as e:
    print(f"Blocked: {e.tool_name}")
    print(f"Verdict: {e.verdict}")           # BLOCK or ESCALATE
    print(f"Decision ID: {e.decision_id}")   # For audit reference
    print(f"Risk score: {e.risk_score}")     # 0.0–1.0
    print(f"Policies: {e.blocking_policies}")  # Policy IDs that triggered block
    print(f"GPT token: {e.governance_proof_token}")  # Governance Proof Token ID

The governance_proof_token is a cryptographically-signed artefact (ECDSA P-256) that proves governance was active at the moment of the decision. Verify it at GET /v1/verify/proof/{token_id}.

Fail-Open Design

gaas-langchain is designed to fail open: if the GaaS API is unreachable, tool calls proceed normally. This ensures your agent keeps working during network interruptions.

Set raise_on_governance_error=True to fail closed instead:

config = GaaSGovernanceConfig(
    raise_on_governance_error=True,  # Fail closed: error if GaaS unreachable
)

Shadow Mode

Test governance policies without blocking agent actions:

# Use the GaaS shadow endpoint — full pipeline, zero enforcement
config = GaaSGovernanceConfig(
    api_url="https://api.gaas.is",
    api_key="gsk_...",
    agent_id="my-agent",
)

# Append ?mode=shadow to the intent URL in your own govern_tool subclass,
# or use GaaSCallbackHandler with enforce=False for pure observation mode.
handler = GaaSCallbackHandler(config=config, enforce=False)

Examples

• examples/langchain_quickstart.py — ReAct agent with governed tools
• examples/langgraph_quickstart.py — LangGraph workflow with governed nodes

Get a GaaS API Key

curl -X POST https://api.gaas.is/v1/onboarding/quickstart \
  -H "Content-Type: application/json" \
  -d '{"organization_name": "Acme Corp", "contact_email": "you@acme.com"}'

Returns your API key (gsk_...) and a pre-configured governance membrane.

License

Apache 2.0. See LICENSE.

Built on GaaS — AI Agent Governance as a Service.