Janus

System-level security for LLM agents via fine-grained policy enforcement on tool calls.

Janus intercepts every tool call an LLM agent makes and validates it against a security policy before execution — following the principle of least privilege. Policies are defined in JSON (or auto-generated by an LLM) and validated at runtime using JSON Schema restrictions.

Status: Alpha (v0.0.4) — APIs are stable but subject to change.

Documentation | GitHub

Table of Contents

• Janus
  • Table of Contents
  • Features
  • Installation
  • Quick Start
    • JanusAgent Parameters
    • JanusAgent Methods
  • Policy Format
    • Full Format
    • Shorthand Format (conditions only)
    • Policy Rule Fields
    • Condition Schemas
    • Evaluation Logic
  • LLM Providers
  • Built-in Tools
    • File Tools
    • Command Tools
  • Custom Tools
    • ToolParam Fields
  • LLM-Generated Policies
    • Standalone Policy Generation
    • Policy Refinement
  • Framework Adapters
    • LangChain
      • Depth 1 — Convert ToolDef list to secured StructuredTool list
      • Depth 2 — Wrap existing LangChain tools
      • Depth 3 — JanusLangChainAgent (turnkey)
    • Google ADK (Gemini)
      • Depth 1 — Convert ToolDef list to ADK-native types
      • Depth 2 — JanusADKAgent (turnkey)
    • Claude Agent SDK (Claude Code)
      • Primary seam — PreToolUse hook
      • Alternative seam — can_use_tool callback
      • Belt-and-braces — guard_tool_body()
  • Standalone Policy Enforcement
  • Runtime Policy Management
  • Error Handling
  • Project Structure
  • License

Features

• Fine-grained policy enforcement — allow or deny tool calls based on argument-level JSON Schema conditions
• Principle of least privilege — policies restrict agents to only what is needed for a specific task
• Multiple policy sources — load from a JSON file, a Python dict, or auto-generate with an LLM
• LLM-generated policies — automatically infer minimum-privilege policies from a user's query
• Policy refinement — incrementally tighten policies as the agent discovers information during a task
• Graph-based capability surfacing — SpiceDB-backed ReBAC and runtime taint tracking (IPI defence) via the integrated PDE engine in janus/policy/pde/
• Built-in tools — ready-to-use file system and command execution tools with workspace sandboxing
• Custom tools — define your own tools with ToolDef / ToolParam; Janus guards them automatically
• 10+ LLM providers — OpenAI, Anthropic, Google Gemini, Azure OpenAI, AWS Bedrock, Ollama, vLLM, Together AI, OpenRouter
• Framework adapters — plug Janus enforcement into LangChain, Google ADK, and Claude Agent SDK (Claude Code) agents
• Standalone enforcer — use PolicyEnforcer independently in any agentic framework
• Three fallback actions — raise PolicyViolation, call sys.exit, or prompt the user interactively
• Workspace isolation — file tools are scoped to a directory; path-traversal attempts are rejected

Installation

Requires Python ≥ 3.10. uv is the recommended package manager.

Core (standalone PolicyEnforcer + OpenAI / OpenAI-compatible providers):

uv add janus-guard

The core install has no SpiceDB/authzed or web-server dependencies, so from janus.policy import PolicyEnforcer works on its own for use as a lightweight tool-call gate.

With optional provider / feature extras:

uv add "janus-guard[pde]"         # SpiceDB-backed ReBAC + taint tracking (PDEEnforcer)
uv add "janus-guard[server]"      # Example web demo UI (FastAPI + uvicorn)
uv add "janus-guard[anthropic]"   # Anthropic Claude
uv add "janus-guard[google]"      # Google Gemini
uv add "janus-guard[bedrock]"     # AWS Bedrock
uv add "janus-guard[langchain]"   # LangChain adapter
uv add "janus-guard[adk]"         # Google ADK adapter
uv add "janus-guard[claude]"      # Claude Agent SDK (Claude Code) adapter
uv add "janus-guard[all]"         # Everything

For development:

uv add "janus-guard[dev]"         # pytest, ruff, mypy

Install from source:

git clone https://github.com/Agentic-AI-Risk-Mitigation/Janus.git
cd janus
uv pip install -e .

Quick Start

from janus import JanusAgent

agent = JanusAgent(
    model="openai/gpt-4o",
    api_key="sk-...",                    # or set OPENAI_API_KEY env var
    use_builtin_tools=True,
    policy="policies.json",
    system_prompt="You are a helpful coding assistant.",
)

response = agent.run("List the Python files in the project.")
print(response)

JanusAgent Parameters

JanusAgent Methods

Policy Format

Policies are JSON documents that map tool names to lists of rules. Each rule specifies whether to allow or deny a tool call and can include argument-level restrictions.

Full Format

{
    "read_file": [
        {
            "priority": 1,
            "effect": 0,
            "conditions": {
                "file_path": {
                    "type": "string",
                    "pattern": "^reports/.*\\.csv$"
                }
            },
            "fallback": 0
        }
    ],
    "run_command": [
        {
            "priority": 1,
            "effect": 1,
            "conditions": {},
            "fallback": 0
        }
    ]
}

Shorthand Format (conditions only)

When you only need to restrict argument values, the shorthand skips priority, effect, and fallback (defaulting to allow, priority 1, raise on violation):

{
    "read_file": {
        "file_path": { "type": "string", "pattern": "^data/.*" }
    }
}

Policy Rule Fields

Condition Schemas

Conditions follow JSON Schema syntax. Common patterns:

{ "type": "string", "pattern": "^/safe/path/.*" }
{ "type": "string", "enum": ["ls", "pwd", "cat"] }
{ "type": "integer", "minimum": 0, "maximum": 100 }
{ "type": "array", "items": { "type": "string" } }

Evaluation Logic

1. Rules for a tool are evaluated in ascending priority order.
2. Allow rule (effect=0): if all conditions pass → tool is allowed immediately.
3. Deny rule (effect=1): if all conditions match → tool is blocked using the configured fallback.
4. If no rule matches → the tool is blocked by default.
5. Tools not listed in the policy are blocked when a policy is loaded.

LLM Providers

Use the "<provider>/<model-name>" format for the model parameter:

Example — Anthropic:

agent = JanusAgent(
    model="anthropic/claude-3-5-sonnet-20241022",
    policy="policies.json",
)

Example — Ollama (local):

agent = JanusAgent(
    model="ollama/llama3.2",
    policy="policies.json",
)

Example — Azure OpenAI:

agent = JanusAgent(
    model="azure/my-deployment",
    api_key="...",
    base_url="https://my-resource.openai.azure.com/",
    api_version="2024-02-01",
    policy="policies.json",
)

Built-in Tools

When use_builtin_tools=True (the default), Janus registers these tools automatically:

File Tools

All file tools are scoped to the workspace directory — attempts to access paths outside the workspace are rejected.

Command Tools

Setting a workspace:

agent = JanusAgent(
    model="openai/gpt-4o",
    workspace="./my_project",   # file tools are sandboxed here
    policy="policies.json",
)

Custom Tools

Define your own tools with ToolDef and ToolParam:

from janus import JanusAgent, ToolDef, ToolParam

def search_database(query: str, limit: int = 10) -> str:
    # your implementation
    return f"Results for '{query}' (limit={limit})"

agent = JanusAgent(
    model="openai/gpt-4o",
    use_builtin_tools=False,    # disable built-ins if not needed
    tools=[
        ToolDef(
            name="search_database",
            description="Search the internal database for records matching a query.",
            params=[
                ToolParam("query", "string", "Search query string"),
                ToolParam("limit", "integer", "Maximum number of results", required=False, default=10),
            ],
            handler=search_database,
        )
    ],
    policy={
        "search_database": [
            {
                "priority": 1,
                "effect": 0,
                "conditions": {
                    "limit": {"type": "integer", "maximum": 50}
                },
                "fallback": 0,
            }
        ]
    },
)

response = agent.run("Find records about renewable energy.")

ToolParam Fields

LLM-Generated Policies

Janus can automatically generate minimum-privilege policies by asking an LLM to infer what tools and restrictions are needed for a given user query:

agent = JanusAgent(
    model="openai/gpt-4o",
    policy="generate",                        # generate on first run()
    policy_model="openai/gpt-4o-2024-08-06", # model used for generation
)

# Policy is generated from this query on the first call
response = agent.run("Read the file sales_2024.csv and summarize the totals.")

# Inspect and save the generated policy
print(agent.get_policy())
agent.save_policy("generated_policy.json")

Standalone Policy Generation

Use generate_policy directly without a full agent:

from janus import generate_policy, save_policy

tools = [
    {"name": "read_file", "description": "Read a file", "args": {...}},
    {"name": "run_command", "description": "Run a shell command", "args": {...}},
]

policy = generate_policy(
    query="Read the quarterly report and list the top 5 expenses.",
    tools=tools,
    model="gpt-4o-2024-08-06",
    manual_confirm=True,   # ask before applying
)

save_policy(policy, "policies.json")

Policy Refinement

After an information-gathering tool call, tighten the policy using discovered values:

from janus import refine_policy

updated = refine_policy(
    query="Send the invoice to the customer.",
    tools=tools,
    tool_call_params={"file_path": "invoices/inv_001.txt"},
    tool_call_result="Customer email: alice@example.com",
    current_policy=current_policy,
    model="gpt-4o-2024-08-06",
    manual_confirm=True,
)

Framework Adapters

LangChain

Three integration depths are available.

Install:

uv add "janus-guard[langchain]"

Depth 1 — Convert ToolDef list to secured StructuredTool list

Use when you build your own LangChain agent but want Janus-guarded tools:

from janus.adapters.langchain import secure_langchain_tools
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI

lc_tools = secure_langchain_tools(my_janus_tools, "policies.json")

llm = ChatOpenAI(model="gpt-4o")
agent = create_tool_calling_agent(llm, lc_tools, prompt)
executor = AgentExecutor(agent=agent, tools=lc_tools)

Depth 2 — Wrap existing LangChain tools

Use when you have an existing LangChain codebase and want to retrofit Janus enforcement:

from janus.adapters.langchain import wrap_langchain_tools

# existing_tools is your existing list of LangChain BaseTool objects
existing_tools = wrap_langchain_tools(existing_tools, "policies.json")
# Pass to your existing AgentExecutor as usual

Depth 3 — JanusLangChainAgent (turnkey)

from janus.adapters.langchain import JanusLangChainAgent

agent = JanusLangChainAgent(
    model="openai/gpt-4o",
    tools=my_janus_tools,
    policy="policies.json",
    system_prompt="You are a helpful assistant.",
    max_iterations=10,
)

response = agent.run("Summarize the quarterly results.")
agent.clear_history()

Google ADK (Gemini)

Two integration depths are available.

Install:

uv add "janus-guard[adk]"

Depth 1 — Convert ToolDef list to ADK-native types

Use when you manage your own Gemini chat loop:

from janus.adapters.adk import secure_adk_tools
from google import genai
from google.genai import types

declarations, handlers = secure_adk_tools(my_janus_tools, "policies.json")

client = genai.Client(api_key="...")
config = types.GenerateContentConfig(
    tools=[types.Tool(function_declarations=declarations)],
    system_instruction="You are helpful.",
    automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
)
chat = client.chats.create(model="gemini-2.0-flash", config=config)

response = chat.send_message("List the files in the project.")
while response.function_calls:
    fc = response.function_calls[0]
    result = handlers[fc.name](**dict(fc.args))
    response = chat.send_message(
        types.Part.from_function_response(fc.name, {"result": result})
    )
print(response.text)

Depth 2 — JanusADKAgent (turnkey)

from janus.adapters.adk import JanusADKAgent

agent = JanusADKAgent(
    model="gemini-2.0-flash",
    tools=my_janus_tools,
    policy="policies.json",
    system_prompt="You are a helpful assistant.",
    max_tool_iterations=10,
)

response = agent.run("What Python files are in the workspace?")
agent.clear_history()

Claude Agent SDK (Claude Code)

The Claude Agent SDK (claude-agent-sdk) runs the tool loop inside the claude CLI subprocess, so Janus cannot sit in the call path the way it does when it owns the loop. Instead the adapter plugs a PolicyEnforcer into the SDK's pre-execution seams, so a policy decision still runs before a tool executes.

Install (also needs the claude CLI on PATH):

uv add "janus-guard[claude]"

Primary seam — PreToolUse hook

Use janus_hooks() to get a ready hooks= dict for ClaudeAgentOptions. A PreToolUse hook fires for every tool call, even ones in allowed_tools and under permission_mode="dontAsk" — so the policy cannot be bypassed:

from claude_agent_sdk import ClaudeAgentOptions, query, create_sdk_mcp_server
from janus.adapters.claude_agent_sdk import janus_hooks

# Policy keyed on bare tool names; the adapter strips the mcp__<server>__ prefix.
TOOL_POLICY = {
    "web_search": [(1, 0, {"query": {"type": "string", "maxLength": 400}}, 0)],
    "fetch_page": [(1, 0, {"url": my_ssrf_check}, 0)],   # callable condition
}

options = ClaudeAgentOptions(
    mcp_servers={"research": create_sdk_mcp_server(name="research", tools=[...])},
    allowed_tools=["mcp__research__web_search", "mcp__research__fetch_page"],
    permission_mode="dontAsk",
    output_format={"type": "json_schema", "schema": SCHEMA},
    # required_args closes Janus's absent-argument bypass (present + non-empty).
    hooks=janus_hooks(TOOL_POLICY, required_args={"fetch_page": ["url"]}),
)

On a policy block the hook returns a permissionDecision: "deny" whose reason is fed back to the model, so the agent can adjust rather than crash. Two SDK-specific behaviours the adapter handles for you:

• StructuredOutput passthrough — when you set output_format, the SDK delivers the final result via an internal StructuredOutput tool call the hook also sees. A default-deny policy would block it and null the result. Such SDK-internal tools are passed through (configurable via passthrough_tools).
• Name mapping — an in-process @tool("fetch", …) on a server named research is invoked as mcp__research__fetch. The adapter maps it back to the bare policy key (fetch). Pass resolve_name=lambda n: n if your policy uses the full prefixed names.

Alternative seam — can_use_tool callback

make_can_use_tool() builds a can_use_tool callback. It is bypassable: the SDK auto-approves any whole-tool allowed_tools entry (and everything under permission_mode="bypassPermissions") before the callback runs, and emits a CanUseToolShadowedWarning. Prefer the hook; use the callback only when nothing shadows it.

Belt-and-braces — guard_tool_body()

Wrap an in-process @tool body so enforcement also runs at execution time, independent of any SDK permission seam:

from janus.adapters.claude_agent_sdk import guard_tool_body

guarded = guard_tool_body("fetch_page", my_async_body, TOOL_POLICY,
                          required_args={"fetch_page": ["url"]})

Standalone Policy Enforcement

PolicyEnforcer can be used independently in any agentic framework — just call enforce() before executing any tool:

from janus.policy import PolicyEnforcer
from janus import PolicyViolation

enforcer = PolicyEnforcer()
enforcer.load("policies.json")

# Before executing a tool:
try:
    enforcer.enforce("read_file", {"file_path": "data/report.csv"})
    result = read_file("data/report.csv")   # proceeds normally
except PolicyViolation as exc:
    print(f"Blocked: {exc.reason}")

# Programmatic policy updates:
enforcer.allow_tools(["list_directory"])    # unconditional allow
enforcer.block_tools(["run_command"])       # unconditional deny
enforcer.update({"write_file": [(1, 0, {}, 0)]})  # merge additional rules

Runtime Policy Management

Policies can be inspected and modified after the agent is created:

# Load a new policy
agent.set_policy("new_policies.json")
agent.set_policy({"read_file": [{"priority": 1, "effect": 0, "conditions": {}, "fallback": 0}]})

# Inspect the current policy
print(agent.get_policy())

# Save to disk
agent.save_policy("saved_policy.json")

# Allow / block specific tools at runtime
agent.allow_tools(["list_directory", "read_file"])
agent.block_tools(["run_command"])

# Manage tools
agent.add_tool(my_new_tool)
agent.remove_tool("old_tool")
print(agent.list_tools())

Error Handling

All Janus exceptions inherit from JanusError:

from janus import (
    JanusError,
    PolicyViolation,         # tool call blocked by policy
    ArgumentValidationError, # argument failed JSON Schema check
    PolicyLoadError,         # policy file not found or invalid JSON
    PolicyGenerationError,   # LLM-based generation failed
    ToolNotFoundError,       # tool name not registered
    ProviderError,           # LLM provider error
)

try:
    response = agent.run("Delete all log files.")
except PolicyViolation as exc:
    print(f"Tool '{exc.tool_name}' was blocked.")
    print(f"Reason: {exc.reason}")
    print(f"Arguments: {exc.arguments}")
except JanusError as exc:
    print(f"Janus error: {exc}")

Project Structure

janus/
├── agent.py                  # JanusAgent — main entry point
├── exceptions.py             # Custom exception classes
├── logger.py                 # Structured logging utilities
├── __init__.py               # Public API re-exports
│
├── llm/
│   ├── base.py               # BaseLLMProvider interface
│   ├── runner.py             # LLMRunner — conversation loop
│   ├── response_types.py     # Provider response types
│   └── providers/
│       ├── openai_provider.py
│       ├── anthropic_provider.py
│       ├── google_provider.py
│       ├── azure_provider.py
│       ├── bedrock_provider.py
│       ├── ollama_provider.py
│       ├── vllm_provider.py
│       ├── together_provider.py
│       └── openrouter_provider.py
│
├── policy/
│   ├── enforcer.py           # PolicyEnforcer — JSON Schema rule engine
│   ├── pde_enforcer.py       # PDEEnforcer — adapter for SpiceDB/taint engine
│   ├── pde/                  # SpiceDB-backed ReBAC + taint (PDE)
│   │   ├── config.py         # SCHEMA, TOOL_TAINT_LIMIT, RISK_TO_TAINT
│   │   ├── interceptor.py    # GraphInterceptor — taint gate + SpiceDB ACL
│   │   ├── discovery.py      # GraphDiscoveryEngine
│   │   └── bootstrap.py     # make_client, bootstrap, Session, allow_tool
│   ├── generator.py          # LLM-based policy generation & refinement
│   ├── loader.py             # JSON parsing and policy persistence
│   └── validator.py          # JSON Schema argument validation
│
├── tools/
│   ├── base.py               # ToolDef, ToolParam dataclasses
│   ├── registry.py           # ToolRegistry — manages registered tools
│   └── builtin/
│       ├── file_tools.py     # read_file, write_file, edit_file, list_directory
│       └── command_tools.py  # run_command, fetch_url
│
└── adapters/
    ├── _base.py              # Shared adapter utilities
    ├── langchain.py          # LangChain integration
    ├── adk.py                # Google ADK (Gemini) integration
    └── claude_agent_sdk.py   # Claude Agent SDK (Claude Code) integration

examples/                     # Demo scenario framework + FastAPI web app + docker-compose.yml for SpiceDB

Running the Demo Web App

The demo web UI lives in examples/app.py. Use the repo-root helper script if you want a single command that starts the FastAPI app without remembering the uvicorn import path.

# Install demo/example dependencies from the repo checkout
uv sync --extra langchain --extra dev
# or: uv sync --extra all --extra dev

# Start the web UI on http://127.0.0.1:8000
./scripts/run_demo_webapp.sh

# Start the web UI and bring up SpiceDB for PDE scenarios
./scripts/run_demo_webapp.sh --with-spicedb

The script also supports --host, --port, and --no-reload.

Running PDE (SpiceDB) Demos

The PDE engine requires a running SpiceDB instance. From a source checkout, install the demo dependencies first; the example runner imports LangChain. Then use the demo app's Docker setup:

Prerequisites: Docker installed and running.

# Install demo/example dependencies from the repo checkout
uv sync --extra langchain --extra dev
# or: uv sync --extra all --extra dev

# Optional smoke test without SpiceDB
uv run python -m examples.run coding_agent_poisoned_readme --protected

# Start SpiceDB (from project root)
cd examples && docker compose up -d && cd ..

# Run the coding-agent taint cascade with PDE — requires SpiceDB
uv run python -m examples.run coding_agent_taint_cascade --protected

# Stop SpiceDB when done
cd examples && docker compose stop && cd ..

What the coding-agent taint cascade exercises:

• ACL-granted tools (readonly, developer roles) pass at low taint
• Python taint gate blocks tools when current_taint > TOOL_TAINT_LIMIT[tool]
• After fetch_url (medium risk), taint rises; git_push (limit 20) is blocked
• Full IPI scenario: agent reads external content → taint increases → dangerous tools blocked

License

MIT