# Galvanize in your agent

Check user input and tool results with Galvanize from LangChain, LlamaIndex or the OpenAI Agents SDK, using a free API key.

Source: https://usezn.com/docs/api/galvanize-agents/

> **This page is for developers who call the optional zn Cloud API from their own code.** Just want to protect your coding agent? Open your own terminal (not the agent's chat), run `npx -y zn-gate init`, then `npx -y zn-gate check`. Step by step: [Install](https://usezn.com/docs/install/).

Galvanize 2 is the model behind the zn Cloud API. Your agent sends it a piece of text (a user message, a retrieved document, a tool result) and gets back `allow` or `block` before the text goes any further. Each integration below is a few lines around one call to `POST https://api.usezn.com/analyze`. The same endpoint is also available as `/v30/analyze`.

You need a key. A Free account is enough: create one in the [dashboard](https://usezn.com/dashboard/), copy a key (it starts with `zn_`) and keep it in an environment variable, not in your code.

The API receives the text exactly as you send it. zn-gate removes the secrets it detects on your machine before anything goes to the cloud; these snippets do not, and they send retrieved documents and tool results as they are. Do not send keys, passwords or personal data, or redact them first, for example with `redact_secrets` from the zn-gate Python package ([Secret redaction](https://usezn.com/docs/zn-gate/redaction/)). What the cloud keeps: [Privacy Policy](https://usezn.com/privacy/).

Want checks that run on your machine, free and offline? Use the local integrations instead: [LangChain, CrewAI and LlamaIndex](https://usezn.com/docs/integrations/frameworks/).

```bash
export ZN_API_KEY=zn_live_your_key
```

## One helper, shared by all three

```python
import os, requests

def analyze(text: str) -> dict:
    r = requests.post("https://api.usezn.com/analyze", json={"input": text}, timeout=12,
                      headers={"Authorization": "Bearer " + os.environ["ZN_API_KEY"]})
    r.raise_for_status()  # fail closed: an error stops the text, it does not let it through
    return r.json()
```

`analyze("...")["verdict"]` is `"block"` or `"allow"`. The full response is in [POST /analyze](https://usezn.com/docs/api/). Use a timeout of 10 to 12 seconds: inputs of more than 2,000 characters are scored in windows and can take longer.

## LangChain

Tested with `langchain-core` 1.6.12 on Python 3.11. Put a `RunnableLambda` in front of the prompt, or after a tool, to stop a blocked text before the model sees it. The prompt must have a single input variable, because `guard` returns a string:

```python
from langchain_core.runnables import RunnableLambda

def guard(text: str) -> str:
    if analyze(text)["verdict"] == "block":
        raise ValueError("Blocked by Galvanize")
    return text

chain = RunnableLambda(guard) | prompt | llm  # use the same guard on a tool's output
```

Install: `pip install langchain-core requests`.

## LlamaIndex

Tested with `llama-index-core` 0.14.25. A node postprocessor drops retrieved chunks that Galvanize blocks, so a chunk it flags never reaches the model:

```python
from llama_index.core.postprocessor.types import BaseNodePostprocessor

class GalvanizeFilter(BaseNodePostprocessor):
    def _postprocess_nodes(self, nodes, query_bundle=None):
        return [n for n in nodes if analyze(n.get_content())["verdict"] == "allow"]

engine = index.as_query_engine(node_postprocessors=[GalvanizeFilter()])
```

Install: `pip install llama-index-core requests`. Every kept or dropped chunk is one API call (more for chunks over 2,000 characters), so retrieve a small `similarity_top_k`.

## OpenAI Agents SDK

Tested with `openai-agents` 0.24.0. An input guardrail with `run_in_parallel=False` runs before the first agent starts and raises `InputGuardrailTripwireTriggered` when the input is blocked. Without `run_in_parallel=False`, the SDK runs the guardrail at the same time as the agent, and the model can answer or call tools before the block arrives:

```python
import asyncio
import requests
from agents import Agent, GuardrailFunctionOutput, input_guardrail

def last_user_text(items):
    for m in reversed(items):
        if isinstance(m, dict) and m.get("role") == "user":
            c = m.get("content", "")
            return c if isinstance(c, str) else " ".join(p.get("text", "") for p in c if isinstance(p, dict))
    return ""

@input_guardrail(run_in_parallel=False)
async def galvanize(ctx, agent, user_input):
    text = user_input if isinstance(user_input, str) else last_user_text(user_input)
    if not text.strip():
        return GuardrailFunctionOutput(output_info="empty", tripwire_triggered=False)
    try:
        verdict = (await asyncio.to_thread(analyze, text))["verdict"]
    except requests.RequestException:
        return GuardrailFunctionOutput(output_info="zn API error", tripwire_triggered=True)
    return GuardrailFunctionOutput(output_info=verdict, tripwire_triggered=verdict == "block")

agent = Agent(name="Support", instructions="Help the user.", input_guardrails=[galvanize])
```

Install: `pip install openai-agents requests`. Catch `InputGuardrailTripwireTriggered` around `Runner.run(agent, user_message)` and answer with a fixed message. If the zn API cannot be reached or rejects the call, the guardrail raises the same tripwire (fail closed); `exception.guardrail_result.output.output_info` tells the two cases apart. When the input is a list (for example from `to_input_list()`), only the latest user message is checked, so every turn costs the same. For tool results, call `analyze` inside the tool before it returns.

## Limits and prices

Same numbers as [Plans and quotas](https://usezn.com/docs/api/quotas/) and the [pricing page](https://usezn.com/pricing/). Team plans are not listed here: see [Plans and quotas](https://usezn.com/docs/api/quotas/).

| Plan | Price | Calls / month |
| - | - | - |
| Free | $0 | 10,000 |
| Pro | $19 / month, or $190 / year; overage $0.10 per 1,000 calls | 250,000 |
| Scale | $99 / month, or $990 / year | 2,000,000 |

- One call covers up to 2,000 characters of `input`. Each extra 1,000 characters, or part of them, is one more call. The maximum is 20,000 characters per request.
- Every request counts, including retries and blocked ones. Cache the verdict for a text you check twice.
- Errors: `400` (no `input`, or more than 20,000 characters), `401` (wrong or revoked key), `402` with `TRIAL_ENDED` (Team subscription not active), `429` with `MONTHLY_LIMIT_REACHED` (quota used up) or `INSUFFICIENT_QUOTA` (this input needs more calls than are left). See [Errors](https://usezn.com/docs/api/errors/).
- Your helper raises on those errors, so the agent stops instead of passing unchecked text on. For low-impact reads you can catch the error and fall back to the local [zn-gate](https://usezn.com/docs/zn-gate/) rules.
- Free keeps the same models as Pro and Scale: every call runs the rules and Galvanize 2.
- When the model is unavailable the API can still answer `200` with the rules only: `decided_by` then starts with `ml-error` (for example `ml-error-rules-fallback`). If you need the model's opinion, treat that as an error.