Ciphyrs — observability and enforcement for AI agents

See every agent, tool call and model call your application makes, as one connected graph; check every tool call against policy before it runs; and mask PII on its way to the model. Built on OpenTelemetry: the spans are standard OTLP, so it sits beside whatever tracing you already have — and your model calls stay yours, made directly against your provider.

Install

pip install ciphyrs

Not on PyPI yet. The latest published ciphyrs is 2.5 and predates the agent SDK. Until 2.8 is published, install from a checkout of this repository: pip install ./packages/sdk/python.

One command. OpenTelemetry (opentelemetry-api, opentelemetry-sdk, opentelemetry-exporter-otlp-proto-http) and httpx come with it — there is no "do I already have OpenTelemetry?" decision to make first. If you do, the SDK attaches to your existing TracerProvider and leaves your exporters alone.

Python 3.9 through 3.14, including the free-threaded 3.14 build (PEP 703). Each release is tested on all of those interpreters, and the package is installed from its built wheel and imported on each one.

Framework integrations are extras: pip install 'ciphyrs[langchain]', 'ciphyrs[crewai]', 'ciphyrs[all]'.

Quick start — agents

import ciphyrs

ciphyrs.init(api_key="cyp_live_...", project="customer-support")

@ciphyrs.tool(enforce=True)
def issue_refund(order: str, amount: float) -> dict:
    """Checked against policy BEFORE it runs. block -> ToolBlocked, nothing executes."""
    return payments.refund(order, amount)

@ciphyrs.agent("BillingAgent", role="worker")
def billing(question: str) -> str:
    resp = model.generate_content(question)   # your model call, unchanged
    return issue_refund(order="A-1041", amount=12.0)

@ciphyrs.agent("RouterAgent", role="router")
def router(message: str) -> str:
    return billing(message)          # called INSIDE router: RouterAgent -> BillingAgent edge

with ciphyrs.session("conv-7f3a"):   # ties the traces of one conversation together
    router("I was charged twice for order A-1041")

# Both agents above are already in the fleet at start-up: decorators register
# themselves. Add peers=[...] to a decorator, or agents=... to init(), for the
# designed edges before any traffic — see "Where the roster comes from".

ciphyrs.shutdown()                    # flush; needed only in short-lived scripts

What each line buys you:

Rules apply to the message and the reply, not only to tools

On by default, no code needed. Every @agent asks the policy engine about its input before it runs and about its output before that output travels further — for the entry agent that is the customer's message and the reply, and for a nested one it is before the parent sees it. Identical text is checked once per turn, so a router that relays its child's answer unchanged costs one check, not two. A blocking rule you apply to the project stops the turn wherever it is hit.

Until 2.9.0 these ran for the outermost agent only, so a nested specialist's own input and output were never examined. That was a gap, not a design.

block raises InputBlocked or OutputBlocked (both are PolicyBlocked, with .stage, .reason and .decision_id to quote back as a reference). A blocked reply is dropped, never returned. guard_input=False / guard_output=False turn either off; guard_fail_closed=True refuses when the guard is unreachable instead of proceeding.

try:
    reply = router(message)
except ciphyrs.PolicyBlocked as b:
    reply = f"I can't help with that here. Reference {b.decision_id}."

Model calls: call your provider directly

Ciphyrs is not in the path of your model call and does not want to be. You call OpenAI, Vertex, Bedrock or anything else exactly as you do today. Nothing about enforcement depends on how that call is made: the message guards, the PII boundary, the tool gate and quarantine all hang off @agent and @tool.

What a model call adds when it IS traced is per-call telemetry — model, latency, tokens, cost, and the prompt and completion the ingest-time detectors read. Three ways to get it, in the order to try them:

1. OpenTelemetry instrumentation for your provider — no code. This SDK is built on OTel, so instrumentation you install emits model spans that land inside our traces. The server reads all three dialects in the field: OTel GenAI semconv (gen_ai.*), OpenInference (llm.*, input.value — Arize/Phoenix packages for LangChain, LlamaIndex, CrewAI) and OpenLLMetry (traceloop.* — Traceloop packages for LangChain, Haystack, LiteLLM). Install the one for your stack and the spans appear with nothing added to your agent code.

2. ciphyrs.llm(...) — a span you open around your own call, for a provider with no instrumentation, a raw HTTP call, or a custom endpoint:

   with ciphyrs.llm("gemini-2.5-flash", provider="vertex_ai", prompt=q) as call:
       resp = client.models.generate_content(model=..., contents=q)
       call.record(resp)     # completion, tokens and model, read off the response
   

   It times a call it does not make, so it cannot see the prompt or the response unless you hand one over. record(resp) understands google-genai, OpenAI-shaped and Anthropic responses; an unrecognised shape records nothing rather than guessing.

3. Nothing at all. Every security control still applies. You lose the per-call row: no model attribution, no tokens, no cost, and the trace's "LLM output" pane stays empty.

PII: what the model sees

ciphyrs.init(api_key=..., project="support")     # masking is already on

On by default since 3.0. Until then pii defaulted to "off", so an application that never read the flag sent its customers' data to the model in the clear. Turn it off with pii="off" for a process that handles no personal data — masking costs a round trip to the NER service on each agent input and each tool result. Operators can override without a redeploy: CIPHYRS_PII=off.

One switch for the whole process. The outermost @agent masks its input through the Ciphyrs NER service (GLiNER, Presidio, spaCy — hosted by us, not downloaded into your app) before any work happens, so the model sees vault tokens such as [CARD_1] instead of the values.

What that covers, precisely. Everything that crosses the boundary: the customer's message, and every tool result on the way back to the model. What it does not cover is text an agent obtains from somewhere else and hands straight to the model — a database read inside a specialist, a file it opens — because that never passed the boundary. Mask those with ciphyrs.mask(), or return them from a @tool, which masks its result. @tool restores the real values for the tool body and masks the result on the way back. The agent restores its reply for the customer. One vault session per turn, so the same value is the same token everywhere. Spans carry the masked text and the PII counts, never the values.

The trace view shows the whole pipeline for each turn — what the customer said, what the model saw, what the model answered, what the customer received — because the boundary agent also sends its original input and restored reply to Ciphyrs (never to the model). pii_capture_raw=False keeps those out of the trace; the vault still holds them for the session.

Fail-open by default, and this is the one place the SDK prefers availability to protection. If masking is unavailable the turn proceeds unmasked and says so — once in the log, and in selfcheck() for the rest of the process's life. The alternative was worse: with masking now on everywhere, fail-closed would make the NER service a hard dependency of every agent that upgrades, so one outage there stops customers being answered at all. Masking someone's data must not become a new way for their support desk to go down.

Set pii_fail_closed=True to refuse instead — a bank should — or CIPHYRS_PII_FAIL_CLOSED=1 to force it from outside. Nothing else in the SDK is fail-open by default: the tool gate is not, and the output guard is not. The LangChain and CrewAI integrations are not either, because you construct those on purpose rather than inheriting them from an upgrade.

@tool(restore_pii=False) hands a tool the tokens as they are. ciphyrs.mask() / ciphyrs.restore() are the explicit forms for anything you assemble yourself.

Tell the model what the tokens are, and in the same breath tell it never to invent one: bracketed values such as [CARD_1] are protected placeholders — use them exactly as given, and never write one yourself; if you need a value you were not given, ask for it. The second half is not optional. A prompt that teaches the format without forbidding invention gets a model that writes [ACCOUNT_1] when it has no account number, passes it to a tool, and answers with a confident balance for an account nobody identified. Measured in a demo application on 9 September 2026. Since SDK 2.9.0 the tool refuses an argument that is an unresolvable placeholder, so the fabrication stops at the gate — but the prompt is where it should not start.

Logs on the trace

Your existing logging calls are enough. init() attaches a handler that forwards every record written inside a span (INFO and above) to the trace and span it was written in, attributed to the agent. Records outside any span are not shipped; the SDK's and OpenTelemetry's own loggers never are.

log = logging.getLogger("bank.cards")

@ciphyrs.agent("CardsAgent")
def cards(msg):
    log.info("card %s is active", last4)     # appears on this trace, from CardsAgent

ciphyrs.log("…", level="warn", source="tool", **fields) is the explicit form. Policy refusals and exceptions are logged for you. Set init(capture_logging=False) to opt out, log_level=logging.DEBUG to widen.

Where the roster comes from

You rarely need to write one.

• Decorators register themselves. Every @ciphyrs.agent is known the moment it is decorated, at import time, before init(). A process shows all of its agents at start-up with no roster at all. Add peers=[...] to the decorator and the designed edges are there too:

  @ciphyrs.agent("RouterAgent", role="router", peers=["BillingAgent", "FraudAgent"])
  def router(message): ...
  

• Frameworks already wrote the topology down. Pass the framework object and fleet_from() reads agents, roles, tools and edges out of it: a LangGraph compiled graph (nodes and edges), an OpenAI Agents SDK Agent (followed through handoffs), a Google ADK agent (through sub_agents), a CrewAI Crew (agents; edges in task order).

  ciphyrs.init(api_key=..., project="support", agents=graph)      # LangGraph
  ciphyrs.init(api_key=..., project="support", agents=triage_agent) # OpenAI Agents SDK
  

• Or list it yourself, for hand-written orchestration that dispatches by name and cannot be read from the code:

  ciphyrs.init(api_key=..., project="support", agents=[
      {"name": "RouterAgent",  "role": "router", "peers": ["BillingAgent"]},
      {"name": "BillingAgent", "role": "worker", "tools": ["issue_refund"]},
  ])
  

  An explicit entry wins over a decorator's on the same name.

What none of these can do is discover an edge from telemetry before it has happened: spans record what did happen. The designed graph is the declaration; observed edges light up on top of it as traffic flows.

Two rules that decide whether the dashboard is useful:

1. Call peers from inside the caller. An edge A → B exists because a span of B has a span of A as its parent. Call billing() after router() has returned and you get two disconnected boxes.
2. fail_closed=True for tools you cannot undo. By default an unreachable guard lets the tool run and marks the span unenforced. For a wire transfer or a kubectl delete, ask for a refusal instead.

Everything is async-safe: async def agents and tools are wrapped the same way, and OpenTelemetry's context carries the nesting across asyncio tasks.

Already running OpenTelemetry? You may not need the SDK at all

For monitoring only, point your existing exporter at Ciphyrs:

export OTEL_EXPORTER_OTLP_ENDPOINT=https://www.ciphyrs.com/v1/otlp
export OTEL_EXPORTER_OTLP_HEADERS=x-api-key=cyp_live_...
export OTEL_SERVICE_NAME=my-agent

Spans carrying the GenAI semantic conventions (gen_ai.agent.name, gen_ai.tool.name, gen_ai.request.model, …), OpenInference or OpenLLMetry attributes are mapped as-is. The SDK adds what OTLP cannot carry: enforcement before the call, and a verified agent identity on it.

Quick start — PII masking

from ciphyrs import CiphyrsClient

client = CiphyrsClient(
    api_key="cyp_live_...",
    base_url="https://www.ciphyrs.com"
)

# Mask PII
result = client.mask("Contact Praveen at 9876125640 and praveen@acme.com")
print(result.masked_text)
# "Contact XXXX_a1b2c3 at XXXX_d4e5f6 and XXXX_g7h8i9"

# Restore PII
restored = client.restore(result.masked_text, result.session_id)
print(restored.restored_text)
# "Contact Praveen at 9876125640 and praveen@acme.com"

One-Shot Protect (recommended)

protect() wraps mask → LLM call → restore so you can't accidentally ship [PERSON_1] to end users:

from openai import OpenAI
oai = OpenAI()

result = client.protect(
    user_message,
    lambda masked, ctx: oai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": masked}],
    ).choices[0].message.content,
)
return result["output"]    # "Hello John, ..." — never "[PERSON_1]"

Active Blocking — Guard (V58)

Inline <50ms allow/block decision, 5 detection layers:

# Option 1 — manual check
guard = client.guard_check(input=user_msg, agent_name="support-bot")
if guard["decision"] == "block":
    return {"error": guard["reason"]}, 400

# Option 2 — full wrap
result = client.guard_wrap(user_msg, lambda inp:
    oai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": inp}],
    ).choices[0].message.content
)
if result["blocked"]:
    return {"error": result["reason"]}, 400
return {"reply": result["output"]}

Security Operations (V55-V59)

# List recent attacks
detections = client.list_detections(days=7, severity="high")

# Plant a honeypot canary
created = client.create_canary(name="fake-admin-secret", scope="output")
print(created["token"])   # save this; plant it where it shouldn't appear

# Generate compliance evidence PDF
report = client.generate_prod_report(
    title="Q4 SOC 2 Evidence",
    range_start="2026-10-01",
    range_end="2026-12-31",
    sections=["security_incidents", "pii_detections", "performance", "cost"],
)
pdf_bytes = client.download_prod_report_pdf(report["report"]["id"])
open("soc2-evidence.pdf", "wb").write(pdf_bytes)

# Share with auditors (no login required)
share = client.share_prod_report(report["report"]["id"], ttl_days=30)
print(share["share_url_path"])

Authentication

All API calls require an x-api-key header. Get your API key from the Ciphyrs Dashboard.

1. Register at ciphyrs.com/register
2. Go to Settings > API Keys
3. Click Generate API Key
4. Copy the key (starts with cyp_live_)

LangChain, CrewAI and any other framework

No framework-specific package is needed, and using one gets you less. The decorators wrap functions, so they work with any framework:

import ciphyrs
ciphyrs.init(api_key="cyp_live_...", project="support", pii="mask")

@ciphyrs.tool(enforce=True)                  # the tool gate: 16 ordered checks
def issue_refund(order: str, amount: float) -> dict:
    return payments.refund(order, amount)

@ciphyrs.agent("Router", role="router")      # PII boundary, message guards, quarantine
def handle(message: str) -> str:
    return my_chain.invoke({"input": message})   # LangChain, CrewAI, anything

@agent on the entry point is what opens the vault session, runs the input and output guards, and refuses to run a quarantined agent. @tool is what the gate sees. Neither knows the framework is there.

Framework-owned tools: the decorator order is load-bearing

Your tools are usually not plain functions — they are LangChain StructuredTools or CrewAI tool objects. Ours must be the inner decorator, so it wraps the callable before the framework takes it:

@langchain_tool                 # framework wraps the governed function
@ciphyrs.tool(enforce=True)     # we wrap the raw callable
def issue_refund(order: str, amount: float) -> dict: ...

Reversed, the framework hands the model the raw callable, our wrapper is never in the call path, and the gate is never asked — silently, because the span still records the call. For a tool object the framework already built, use govern() instead and the order stops mattering:

tool = StructuredTool.from_function(issue_refund)
ciphyrs.govern(tool)            # same object, now behind the gate

The roster, with nothing written by hand

ciphyrs.init(api_key="...", agents=ciphyrs.fleet_from(my_crew))

fleet_from() reads the designed topology out of a LangGraph compiled graph, an OpenAI Agents Agent (through handoffs), a Google ADK agent (through sub_agents), a CrewAI Crew (edges in task order), or anything with .agents. A plain LangChain AgentExecutor yields one agent carrying its tool names, because a chain is one agent that holds tools — not a team.

Spans inside a chain you did not write

See "Model calls" above: install the OpenTelemetry instrumentation for your framework or provider and its spans land inside our traces. Nothing of ours is required, and nothing of ours should be written for it.

Is it actually on?

state = ciphyrs.selfcheck()
if state["problems"]:
    log.error("Ciphyrs is not protecting this process: %s", state["problems"])

handle.enforcing only means "there is a key". selfcheck() reports what is genuinely live — agents registered, tools governed, masking, guards, propagation, rules that failed to compile locally — and names each problem in words. It makes no network call, so it is safe in a readiness probe.

The older PII-only integrations

ciphyrs.integrations.langchain and ciphyrs.integrations.crewai (CiphyrsPIICallback, CiphyrsShield, CiphyrsCrewShield, …) are deprecated. They do PII masking and restoring and nothing else — no message guards, no tool gate, no quarantine, no fleet declaration, no detection layers. They are kept working for existing users; new applications should use the decorators above, which are less code and the whole platform.

Async Support

Declare the function async def and you are done. @agent and @tool detect a coroutine and install async wrappers, and every platform call they make — masking, restore, the message guards, the tool gate — runs off your event loop.

@ciphyrs.tool(enforce=True, fail_closed=True)
async def issue_refund(order: str, amount: float) -> dict:
    return await payments.refund(order, amount)

@ciphyrs.agent("BillingAgent", role="worker")
async def billing(message: str) -> str:
    resp = await client.chat(message)              # your provider, awaited
    return await issue_refund(order="A-1041", amount=12.0)

Why it matters. The HTTP client is httpx.Client — a blocking socket read. In a synchronous application that is correct and invisible. Inside a coroutine it stops the whole loop, not just that coroutine, so one agent's tool check stalls every other request the process is serving. Measured against a 0.5 s endpoint before this was fixed: four concurrent guarded calls took 10.62 s wall clock and the loop ticked zero times where a free loop would have ticked about a thousand.

A synchronous agent called from a coroutine still blocks

A def cannot await, so the SDK cannot offload it. If you call one from inside a running loop it will block that loop, and the SDK says so once:

[ciphyrs] @agent 'BillingAgent' is a synchronous function called from inside an
event loop: its Ciphyrs calls will block that loop, stalling every other request
in this process. Declare it `async def` (the SDK offloads automatically), or call
it through asyncio.to_thread() / starlette's run_in_threadpool().

selfcheck() reports it too, naming the functions. Either fix works:

@app.post("/chat")
async def chat(body: ChatIn):
    return await run_in_threadpool(lambda: route(body.message))   # sync agent

Do not pass an async client to init()

init(client=AsyncCiphyrsClient(...)) raises. The fleet declaration, heartbeat, server defaults and rule refresh are synchronous, and an async client makes all four silently return un-awaited coroutines. You do not need it — an async def agent already keeps the loop free.

The standalone client

Separate from the agent SDK, for direct PII calls with no init():

from ciphyrs import AsyncCiphyrsClient

async with AsyncCiphyrsClient(api_key="cyp_live_...") as client:
    result = await client.mask("Contact Praveen at praveen@acme.com")
    restored = await client.restore(result.masked_text, result.session_id)

Detected Entity Types

The pre-OpenTelemetry tracer (ciphyrs.agenttrace)

CiphyrsTracer predates the OpenTelemetry-based API above and remains for integrations already built on it. New code should use ciphyrs.init() and the decorators; cross-process propagation there is OpenTelemetry's own (W3C traceparent), which every instrumented HTTP client and server already speaks. What follows applies to CiphyrsTracer.

An agent running on your own infrastructure — Oracle, AWS, Azure, GCP, on-prem — needs nothing from us but outbound HTTPS and an API key.

Agents in separate processes appear as one connected system

The topology graph is derived: an edge A → B exists because a span of agent B names a span of agent A as its parent. Inside one process the SDK tracks that for you. Across processes it used to need hand-threaded headers, and without them two agents that talked constantly rendered as two disconnected dots.

Outbound httpx and requests calls made inside a span now carry W3C traceparent and baggage automatically:

from ciphyrs.agenttrace import CiphyrsTracer, TraceConfig

tracer = CiphyrsTracer(TraceConfig(project="orders", agent_name="RouterAgent"))

with tracer.trace("handle order") as t:
    with t.span("RouterAgent", kind="agent"):
        # headers are added for you — nothing to pass
        httpx.post("https://billing.internal/charge", json=payload)

On the receiving side, one line activates the caller's trace for the request:

from fastapi import FastAPI
from ciphyrs.propagation import CiphyrsASGIMiddleware

app = FastAPI()
app.add_middleware(CiphyrsASGIMiddleware)      # Flask: instrument_flask(app)

@app.post("/charge")
def charge():
    with tracer.trace("charge") as t:          # continues the caller's trace
        with t.span("BillingAgent", kind="agent"):
            ...

That is all the edge needs. t.is_continuation is True and the first span is parented to the caller's span. Django and other WSGI apps use CiphyrsWSGIMiddleware.

Queues, gRPC, or a framework not listed — two functions:

from ciphyrs.propagation import inject, remote_context

queue.send(body, headers=inject({}))              # producer

with remote_context(message.headers):             # consumer
    with tracer.trace("handle job") as t:
        with t.span("Worker", kind="agent"):
            ...

Peers instrumented with plain OpenTelemetry interoperate: ids are generated in W3C shape (32/16 hex), so a non-Ciphyrs service joins the same trace.

Set propagate=False on TraceConfig (or CIPHYRS_PROPAGATE=false) to opt out. Calls to the Ciphyrs API itself are never decorated.

Spans nest without bookkeeping

A span opened inside another is its child, so the graph has edges even within one process:

with tracer.trace("run") as t:
    with t.span("Router", kind="agent"):
        with t.span("Billing", kind="agent"):    # child of Router
            ...

Pass parent= explicitly to override it.

Health is reported, and can be observed

Heartbeats are on by default (they used to default to off, so almost no agent ever reported liveness) and carry the interval they beat at, so the platform sizes each agent's "down" window to that agent instead of applying one global threshold to a fleet whose agents beat at very different rates. They also ship process metrics — thread lag, error rate over the spans since the last beat, uptime, and RSS/CPU when psutil is installed — so the fleet can show degraded before down:

tracer = CiphyrsTracer(TraceConfig(
    project="orders",
    agent_name="RouterAgent",     # visible in the fleet before any traffic
    heartbeat_interval=60,        # 0 disables
    heartbeat_metrics=True,
))

To make down something Ciphyrs observed rather than inferred from silence, register a URL the platform polls (dashboard → agent → monitoring, or PATCH /v1/trace/agents/{id}/monitoring). Private, loopback and cloud-metadata addresses are refused by the prober and surface as probe_status: blocked — a misconfiguration, never an outage.

If you cannot propagate headers

The platform also infers edges from shared traces, correlation ids and shared sessions. Those render dashed with a confidence score, and an operator can confirm or dismiss them. A real propagated edge always wins over an inferred one, so adding the middleware above upgrades them automatically.

Links

• Website
• Documentation
• Dashboard