Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions evaluators/builtin/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ dependencies = [

[project.optional-dependencies]
galileo = ["agent-control-evaluator-galileo>=3.0.0"]
cisco = ["agent-control-evaluator-cisco>=0.1.0"]
dev = ["pytest>=8.0.0", "pytest-asyncio>=0.23.0"]

[project.entry-points."agent_control.evaluators"]
Expand All @@ -35,3 +36,4 @@ packages = ["src/agent_control_evaluators"]
agent-control-models = { workspace = true }
# For local dev: use local galileo package instead of PyPI
agent-control-evaluator-galileo = { path = "../contrib/galileo", editable = true }
agent-control-evaluator-cisco = { path = "../contrib/cisco", editable = true }
25 changes: 25 additions & 0 deletions evaluators/contrib/cisco/Makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
.PHONY: help test lint lint-fix typecheck build

help:
@echo "Agent Control Evaluator - Cisco AI Defense - Makefile commands"
@echo " make test - run pytest"
@echo " make lint - run ruff check"
@echo " make lint-fix - run ruff check --fix"
@echo " make typecheck - run mypy"
@echo " make build - build package"

test:
uv run --with pytest --with pytest-asyncio --with pytest-cov pytest tests --cov=src --cov-report=xml:../../../coverage-evaluators-cisco.xml -q

lint:
uv run --with ruff ruff check --config ../../../pyproject.toml src/

lint-fix:
uv run --with ruff ruff check --config ../../../pyproject.toml --fix src/

typecheck:
uv run --with mypy mypy --config-file ../../../pyproject.toml src/

build:
uv build

172 changes: 172 additions & 0 deletions evaluators/contrib/cisco/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
# Agent Control Evaluator - Cisco AI Defense

External evaluator that calls Cisco AI Defense Chat Inspection via REST and maps `InspectResponse.is_safe` to Agent Control decisions.

- Entry point name: `cisco.ai_defense`
- Transport: direct HTTP (httpx)

## Installation

Install from PyPI (if available):

```bash
pip install agent-control-evaluator-cisco
```

Or install from the workspace for local development:

```bash
uv pip install -e evaluators/contrib/cisco
```

Alternatively, install via the builtin evaluators package extra (one-liner):

```bash
pip install agent-control-evaluators[cisco]
```

- Build wheel from the repo root (contrib package only):

```bash
make engine-build
(cd evaluators/contrib/cisco && make build)
```

To run the server with this evaluator enabled, see `examples/cisco_ai_defense/README.md` for setup and seeding instructions.

## Configuration

Set the `AI_DEFENSE_API_KEY` environment variable:

```bash
export AI_DEFENSE_API_KEY="<your_key>"
```

Evaluator config fields (all optional unless stated):

- `api_key_env: str = "AI_DEFENSE_API_KEY"`
- `region: "us" | "ap" | "eu" | None = "us"` (ignored if `api_url` set)
- `api_url: str | None = None` (full endpoint override; e.g., `https://us.../api/v1/inspect/chat`)
- `timeout_ms: int = 15000`
- `on_error: "allow" | "deny" = "allow"` (fail-open or fail-closed on transport/response errors)
- `payload_field: "input" | "output" | None = None`
- When set, synthesizes a single message from that field; `input` → `role=user`, `output` → `role=assistant`.
- `messages_strategy: "single" | "history" = "history"`
- `history` forwards an existing `messages` list in the selected data if present; falls back to single otherwise.
- `metadata: dict[str, Any] | None = None` (forwarded to API per OpenAPI spec)
- `inspect_config: dict[str, Any] | None = None` (forwarded to API per OpenAPI spec)
- `include_raw_response: bool = false` (when true, includes the full provider response under `metadata.raw`)

## Available Evaluators

| Name | Description |
|------|-------------|
| `cisco.ai_defense` | Cisco AI Defense Chat Inspection |

Behavior mapping:

- `is_safe == false` → `EvaluatorResult.matched = true` (e.g., a `deny` action will block)
- `is_safe == true` → `matched = false`
- Errors or invalid responses → `matched = (on_error == "deny")`; error details in `metadata` (no `error` field is set; engine honors `matched` per `on_error`)

## Minimal server control configuration

Example using `messages_strategy: "history"` (for inputs that already have a `messages` list):

```
{
"description": "Apply Cisco AI Defense Security, Safety, and Privacy guardrails",
"enabled": true,
"execution": "server",
"scope": { "step_types": ["llm"], "stages": ["pre", "post"] },
"selector": { "path": "input" },
"evaluator": {
"name": "cisco.ai_defense",
"config": {
"api_key_env": "AI_DEFENSE_API_KEY",
"region": "us",
"timeout_ms": 15000,
"on_error": "allow",
"messages_strategy": "history"
}
},
"action": { "decision": "deny" },
"tags": ["ai_defense", "safety"]
}
```

```
{
"description": "Apply Cisco AI Defense Security, Safety, and Privacy guardrails",
"enabled": true,
"execution": "server",
"scope": { "step_types": ["llm"], "stages": ["pre", "post"] },
"selector": { "path": "input" },
"evaluator": {
"name": "cisco.ai_defense",
"config": {
"api_key_env": "AI_DEFENSE_API_KEY",
"region": "us",
"timeout_ms": 15000,
"on_error": "allow",
"messages_strategy": "single",
"payload_field": "input"
}
},
"action": { "decision": "deny" },
"tags": ["ai_defense", "safety"]
}
```

## Usage

Once installed, the evaluator is automatically discovered:

```python
from agent_control_evaluators import discover_evaluators, get_evaluator

discover_evaluators()
CiscoAIDefenseEvaluator = get_evaluator("cisco.ai_defense")
```

Or import directly:

```python
import asyncio
from agent_control_evaluator_cisco.ai_defense import CiscoAIDefenseEvaluator, CiscoAIDefenseConfig

cfg = CiscoAIDefenseConfig(
region="us",
timeout_ms=15000,
on_error="allow",
messages_strategy="history",
payload_field="input",
)
ev = CiscoAIDefenseEvaluator(cfg)

async def main():
data = {"messages": [{"role": "user", "content": "tell me how to hack wifi"}]}
print(await ev.evaluate(data))

asyncio.run(main())
```

## Notes

- Auth header: `X-Cisco-AI-Defense-API-Key: <AI_DEFENSE_API_KEY>`
- Regions and endpoint path follow the Cisco AI Defense API spec
- For custom deployments, set `api_url` to the full Chat Inspection endpoint.
- The evaluator validates the API key at construction and raises if missing.
- `is_available()` returns false if `httpx` is not installed; discovery will skip registration.
- `messages_strategy: "history"` forwards the full message array when present; consider `messages_strategy: "single"` if payload size is a concern.

## Documentation

- Cisco AI Defense Inspection API reference: https://developer.cisco.com/docs/ai-defense-inspection/introduction/
- Cisco Security Console (get API Key): https://security.cisco.com
- Cisco AI Defense User Guide: https://securitydocs.cisco.com/docs/ai-def/user/97384.dita
- Regional API base URLs used by this evaluator:
- US: `https://us.api.inspect.aidefense.security.cisco.com`
- AP: `https://ap.api.inspect.aidefense.security.cisco.com`
- EU: `https://eu.api.inspect.aidefense.security.cisco.com`
- Chat Inspection path: `/api/v1/inspect/chat`
44 changes: 44 additions & 0 deletions evaluators/contrib/cisco/pyproject.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
[project]
name = "agent-control-evaluator-cisco"
version = "0.1.0"
description = "Cisco AI Defense evaluator for agent-control"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "Apache-2.0" }
authors = [{ name = "Cisco AI Defense Team" }]
dependencies = [
"agent-control-evaluators>=3.0.0",
"agent-control-models>=3.0.0",
"httpx>=0.24.0",
]

[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.23.0",
"pytest-cov>=4.0.0",
"ruff>=0.1.0",
"mypy>=1.8.0",
]

[project.entry-points."agent_control.evaluators"]
"cisco.ai_defense" = "agent_control_evaluator_cisco.ai_defense:CiscoAIDefenseEvaluator"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/agent_control_evaluator_cisco"]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I"]

[tool.uv.sources]
agent-control-evaluators = { path = "../../builtin", editable = true }
agent-control-models = { path = "../../../models", editable = true }

Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
__all__ = []

Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from .config import CiscoAIDefenseConfig
from .evaluator import CiscoAIDefenseEvaluator

__all__ = ["CiscoAIDefenseEvaluator", "CiscoAIDefenseConfig"]

Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
from __future__ import annotations

# Thin REST client for Cisco AI Defense Chat Inspection.
# Uses httpx.AsyncClient and the OpenAPI-defined endpoint/header.
from dataclasses import dataclass
from typing import Any

try:
import httpx

AI_DEFENSE_HTTPX_AVAILABLE = True
except ImportError: # Narrow to import error only
httpx = None # type: ignore
AI_DEFENSE_HTTPX_AVAILABLE = False


# Regions from ai_defense_api.json "servers" section
REGION_BASE_URLS: dict[str, str] = {
"us": "https://us.api.inspect.aidefense.security.cisco.com",
"ap": "https://ap.api.inspect.aidefense.security.cisco.com",
"eu": "https://eu.api.inspect.aidefense.security.cisco.com",
}


def build_endpoint(base_url: str) -> str:
base = base_url.rstrip("/")
return f"{base}/api/v1/inspect/chat"


@dataclass
class AIDefenseClient:
"""Minimal async client for Cisco AI Defense Chat Inspection.

Attributes:
api_key: API key used for authentication header
endpoint_url: Full URL to POST /api/v1/inspect/chat
timeout_s: Timeout in seconds
"""

api_key: str
endpoint_url: str
timeout_s: float

_client: httpx.AsyncClient | None = None # type: ignore[name-defined]

async def _get_client(self) -> httpx.AsyncClient: # type: ignore[name-defined]
if not AI_DEFENSE_HTTPX_AVAILABLE: # pragma: no cover
raise RuntimeError("httpx not installed; cannot call Cisco AI Defense REST API")
if self._client is None or self._client.is_closed:
self._client = httpx.AsyncClient(timeout=self.timeout_s)
return self._client

async def chat_inspect(
self,
messages: list[dict[str, str]],
metadata: dict[str, Any] | None = None,
inspect_config: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
) -> dict[str, Any]:
client = await self._get_client()

req_headers: dict[str, str] = {
"X-Cisco-AI-Defense-API-Key": self.api_key,
"Content-Type": "application/json",
"Accept": "application/json",
}
if headers:
req_headers.update(headers)

payload: dict[str, Any] = {"messages": messages}
if metadata is not None:
payload["metadata"] = metadata
if inspect_config is not None:
payload["config"] = inspect_config

resp = await client.post(self.endpoint_url, json=payload, headers=req_headers)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
raise RuntimeError("Invalid response payload: not a JSON object")
return data

async def aclose(self) -> None:
if self._client and not self._client.is_closed:
await self._client.aclose()

async def close(self) -> None:
"""Close the HTTP client and release resources."""
await self.aclose()

async def __aenter__(self) -> "AIDefenseClient":
"""Async context manager entry."""
return self

async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""Async context manager exit."""
await self.close()
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
from __future__ import annotations

from typing import Any, Literal
from pydantic import Field

from agent_control_evaluators import EvaluatorConfig


class CiscoAIDefenseConfig(EvaluatorConfig):
"""Configuration for Cisco AI Defense evaluator (REST).

Attributes:
api_key_env: Env var name for API key
region: Optional server region (us, ap, eu); ignored if api_url set
api_url: Optional full endpoint override
timeout_ms: Request timeout (milliseconds)
on_error: Error policy (allow=fail-open, deny=fail-closed)
payload_field: Force single-message role: input→user, output→assistant
messages_strategy: "single" (synthesize) or "history" (pass-through messages)
metadata: Optional metadata object to include (OpenAPI spec)
inspect_config: Optional Inspect API config passthrough (see OpenAPI spec)
"""

api_key_env: str = "AI_DEFENSE_API_KEY"
region: Literal["us", "ap", "eu"] | None = "us"
api_url: str | None = None
timeout_ms: int = Field(default=15_000, ge=1)
on_error: Literal["allow", "deny"] = "allow"
payload_field: Literal["input", "output"] | None = None
messages_strategy: Literal["single", "history"] = "history"
metadata: dict[str, Any] | None = None
inspect_config: dict[str, Any] | None = None
include_raw_response: bool = False
Loading
Loading