Galvanize in your agent

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.

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, 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). What the cloud keeps: Privacy Policy.

Want checks that run on your machine, free and offline? Use the local integrations instead: LangChain, CrewAI and LlamaIndex.

export ZN_API_KEY=zn_live_your_key

One helper, shared by all three

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. 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:

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:

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:

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 and the pricing page. Team plans are not listed here: see Plans and quotas.

• 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.
• 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 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.