Metadata-Version: 2.4
Name: agentguard-platform
Version: 1.0.1
Summary: AgentGuard Python SDK — Runtime security for AI agents
Project-URL: Homepage, https://github.com/BLACK0HEART/agentguard
Project-URL: Repository, https://github.com/BLACK0HEART/agentguard
Project-URL: Documentation, https://github.com/BLACK0HEART/agentguard#readme
Author-email: Achraf Boulahya <achrafboulahya00@gmail.com>
License-Expression: Apache-2.0
Keywords: agents,ai,crewai,langchain,monitoring,security
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Security
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27.0
Requires-Dist: opentelemetry-api>=1.24.0
Requires-Dist: opentelemetry-exporter-otlp>=1.24.0
Requires-Dist: opentelemetry-sdk>=1.24.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: structlog>=24.0.0
Provides-Extra: all
Requires-Dist: agentguard[crewai,langchain]; extra == 'all'
Provides-Extra: crewai
Requires-Dist: crewai>=0.28.0; extra == 'crewai'
Provides-Extra: langchain
Requires-Dist: langchain>=0.1.0; extra == 'langchain'
Description-Content-Type: text/markdown

# AgentGuard Python SDK

Runtime security for AI agents — LangChain, CrewAI, and any Python tool.

[![PyPI version](https://badge.fury.io/py/agentguard.svg)](https://badge.fury.io/py/agentguard)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: Apache 2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)

---

## Installation

```bash
pip install agentguard-platform
```

With LangChain support:
```bash
pip install "agentguard-platform[langchain]"
```

With CrewAI support:
```bash
pip install "agentguard-platform[crewai]"
```

---

## Quick Start (5 lines)

```python
from agentguard import AgentGuard

AgentGuard.configure(
    api_url="http://localhost:8000",
    api_key="your-api-key",
    agent_id="your-agent-uuid",
)

@AgentGuard.tool
def read_file(path: str) -> str:
    return open(path).read()
```

Every call to `read_file()` is now:
- Evaluated against your security policies before execution
- Blocked if it matches a threat pattern (path traversal, prompt injection, etc.)
- Logged in the AgentGuard audit trail
- Monitored for risk score and behavioral anomalies

---

## LangChain Integration

### Option A — Wrap existing tools (recommended)

```python
from langchain.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from agentguard import AgentGuard

AgentGuard.configure(api_url="...", api_key="...", agent_id="...")

# Wrap all tools in one call
tools = AgentGuard.protect_langchain_tools([
    DuckDuckGoSearchRun(),
    WikipediaQueryRun(),
])

executor = AgentExecutor(agent=agent, tools=tools)
```

### Option B — Callback handler (zero tool changes)

```python
from langchain.agents import AgentExecutor
from agentguard import AgentGuard

AgentGuard.configure(...)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[AgentGuard.callback_handler()],
)
```

### Option C — @AgentGuard.tool decorator

```python
from agentguard import AgentGuard

AgentGuard.configure(...)

@AgentGuard.tool
def search_web(query: str) -> str:
    return search_api(query)

@AgentGuard.tool
def read_file(path: str) -> str:
    return open(path).read()
```

### Async LangChain

```python
handler = AgentGuard.async_callback_handler()

result = await executor.arun(
    "What is the capital of France?",
    callbacks=[handler],
)
```

### Full LangChain example

```python
import os
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain import hub
from agentguard import AgentGuard

AgentGuard.configure(
    api_url=os.getenv("AGENTGUARD_URL", "http://localhost:8000"),
    api_key=os.getenv("AGENTGUARD_API_KEY"),
    agent_id=os.getenv("AGENTGUARD_AGENT_ID"),
)

def search(query: str) -> str:
    return f"Results for: {query}"

def calculate(expr: str) -> str:
    return str(eval(expr))  # guarded by AgentGuard policies

tools = AgentGuard.protect_langchain_tools([
    Tool(name="search", func=search, description="Search the web"),
    Tool(name="calculator", func=calculate, description="Evaluate math"),
])

llm = ChatOpenAI(temperature=0)
agent = create_react_agent(llm, tools, hub.pull("hwchase17/react"))
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[AgentGuard.callback_handler()],
)

result = executor.invoke({"input": "What is 42 * 7?"})
```

---

## OpenAI Integration

Three patterns — pick the one that fits your architecture.

### Option A — `wrap_openai_tool` (Chat Completions, recommended)

```python
from agentguard import AgentGuardClient
from agentguard.hooks.openai_hook import wrap_openai_tool
from openai import OpenAI

openai_client = OpenAI()
guard = AgentGuardClient(api_key="ag_...", base_url="https://your-agentguard/api/v1")

def search_web(query: str) -> str:
    # your implementation
    return f"results for {query}"

protected_search = wrap_openai_tool(search_web, guard, agent_id="agent-123")

# Use exactly like the original function — blocked calls raise AgentGuardBlockedError
result = protected_search(query="latest news")
```

### Option B — `filter_tool_calls` (Assistants API)

```python
from agentguard.hooks.openai_hook import filter_tool_calls

# Inside your run polling loop:
if run.status == "requires_action":
    tool_calls = run.required_action.submit_tool_outputs.tool_calls
    allowed, blocked = filter_tool_calls(guard, tool_calls, agent_id="agent-123")

    tool_outputs = []
    for b in blocked:
        tool_outputs.append({"tool_call_id": b["tool_call_id"], "output": b["error_message"]})
    for tc in allowed:
        tool_outputs.append({"tool_call_id": tc.id, "output": dispatch(tc)})

    openai_client.beta.threads.runs.submit_tool_outputs(thread_id, run.id, tool_outputs=tool_outputs)
```

### Option C — `AgentGuardOpenAICallback` (manual dispatch loop)

```python
from agentguard.hooks.openai_hook import AgentGuardOpenAICallback

cb = AgentGuardOpenAICallback(client=guard, agent_id="agent-123")

response = openai_client.chat.completions.create(model="gpt-4o", messages=messages, tools=tools)
for tool_call in response.choices[0].message.tool_calls:
    args = json.loads(tool_call.function.arguments)
    cb.on_tool_start(tool_call.function.name, args)  # raises AgentGuardBlockedError if blocked
    result = dispatch(tool_call)
    cb.on_tool_end(tool_call.function.name, result)
```

---

## CrewAI Integration

### Option A — @AgentGuard.crewai_tool decorator

```python
from crewai import Agent, Crew, Task
from agentguard import AgentGuard

AgentGuard.configure(api_url="...", api_key="...", agent_id="...")

@AgentGuard.crewai_tool
def research_topic(topic: str) -> str:
    """Research a topic thoroughly."""
    return search_api(topic)

@AgentGuard.crewai_tool(name="safe_file_writer")
def write_report(content: str, filename: str) -> str:
    """Write a report — AgentGuard checks for path traversal."""
    return write_file(filename, content)
```

### Option B — Wrap existing CrewAI tools

```python
from crewai_tools import SerperDevTool, FileReadTool
from agentguard import AgentGuard

AgentGuard.configure(...)

tools = AgentGuard.protect_crewai_tools([
    SerperDevTool(),
    FileReadTool(),
])
```

### Full CrewAI example

```python
from crewai import Agent, Crew, Task
from agentguard import AgentGuard

AgentGuard.configure(
    api_url="http://localhost:8000",
    api_key="your-api-key",
    agent_id="your-agent-uuid",
)

@AgentGuard.crewai_tool
def search_web(query: str) -> str:
    return search_api(query)

researcher = Agent(
    role="Research Analyst",
    goal="Research topics thoroughly",
    backstory="Expert researcher with broad knowledge.",
    tools=[search_web],
)

task = Task(
    description="Research the latest AI safety developments.",
    expected_output="Comprehensive summary.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()
```

---

## Memory Firewall

Inspect RAG content before feeding it to your agent:

```python
from agentguard import AgentGuard

AgentGuard.configure(...)

def safe_rag_retrieval(query: str, documents: list[str]) -> list[str]:
    """Filter RAG chunks through AgentGuard Memory Firewall."""
    safe_docs = []
    for doc in documents:
        result = AgentGuard.inspect_memory(doc, source="rag")
        if result.safe:
            safe_docs.append(doc)
        else:
            print(f"Blocked poisoned chunk: {result.threats}")
    return safe_docs

# Use in your RAG pipeline
raw_chunks = vector_db.similarity_search(query, k=5)
safe_chunks = safe_rag_retrieval(query, [c.page_content for c in raw_chunks])
# Feed only safe_chunks to your agent
```

With strict mode (raises exception on threat):

```python
from agentguard import AgentGuard, AgentGuardBlockedError

try:
    AgentGuard.inspect_memory(doc, source="rag", raise_on_threat=True)
except AgentGuardBlockedError as e:
    print(f"Poisoned document blocked: {e}")
```

---

## Kill Switch

Check or enforce kill switch status before running agents:

```python
from agentguard import AgentGuard, AgentGuardKillSwitchError

AgentGuard.configure(...)

# Check status
status = AgentGuard.check_kill_switch()
if status.active:
    print(f"Kill switch active: {status.reason}")
    sys.exit(1)

# Or assert (raises if active)
try:
    AgentGuard.assert_kill_switch_inactive()
except AgentGuardKillSwitchError as e:
    print(f"Emergency stop: {e}")
```

Kill switch is automatically checked for all `@AgentGuard.tool` decorated functions — if the kill switch is active, tool calls return `decision=block`.

---

## Error Handling

```python
from agentguard import (
    AgentGuard,
    AgentGuardBlockedError,
    AgentGuardQuarantineError,
    AgentGuardRequireApprovalError,
    AgentGuardKillSwitchError,
    AgentGuardTimeoutError,
    AgentGuardConnectionError,
)

@AgentGuard.tool
def risky_tool(command: str) -> str:
    return execute(command)

try:
    result = risky_tool("ls -la /etc")

except AgentGuardKillSwitchError as e:
    # Global kill switch is active — all agents stopped
    print(f"Kill switch: {e.reason}")

except AgentGuardBlockedError as e:
    # Blocked by a policy rule
    print(f"Blocked by '{e.policy_name}' (risk={e.risk_score:.1f})")
    print(f"Reasons: {e.reasons}")

except AgentGuardQuarantineError as e:
    # Agent quarantined — needs human review
    print(f"Quarantined: {e.tool_call_id}")

except AgentGuardRequireApprovalError as e:
    # Needs explicit human approval before proceeding
    print(f"Approval needed for tool call: {e.tool_call_id}")

except AgentGuardTimeoutError:
    # API timed out AND block_on_error=True
    print("AgentGuard unreachable — blocked for safety")

except AgentGuardConnectionError:
    # API unreachable AND block_on_error=True
    print("Could not reach AgentGuard — blocked for safety")
```

---

## Configuration

### Environment variables (no code change needed)

```bash
export AGENTGUARD_URL="http://localhost:8000"
export AGENTGUARD_API_KEY="your-api-key"
export AGENTGUARD_AGENT_ID="your-agent-uuid"
```

### Programmatic configuration

```python
AgentGuard.configure(
    api_url="http://localhost:8000",     # AgentGuard API URL
    api_key="your-api-key",             # API key or JWT token
    agent_id="your-agent-uuid",         # Agent UUID from AgentGuard
    timeout=5.0,                        # HTTP timeout (seconds)
    block_on_error=True,                # FAIL-SECURE: block if API unreachable
    verify_ssl=True,                    # Verify TLS certificates
    max_retries=2,                      # Retry on 5xx responses
)
```

**`block_on_error=True` (default):** If AgentGuard API is unreachable, tool calls are BLOCKED.
This is the safe default for production.

**`block_on_error=False`:** If AgentGuard API is unreachable, tool calls are ALLOWED.
Only use in development/testing.

---

## Direct Client Usage

For advanced use cases where you need direct API access:

```python
from agentguard import AgentGuardClient

with AgentGuardClient(
    api_url="http://localhost:8000",
    api_key="your-key",
    block_on_error=True,
) as client:
    result = client.evaluate_tool_call(
        agent_id="your-agent-uuid",
        tool_name="read_file",
        tool_input={"path": "/etc/passwd"},
        enforce=True,   # raises AgentGuardBlockedError if blocked
    )
    print(f"Decision: {result.decision}")
    print(f"Risk score: {result.risk_score}")
    print(f"Policy: {result.policy_name}")
```

Async:

```python
from agentguard import AsyncAgentGuardClient

async with AsyncAgentGuardClient(api_url="...", api_key="...") as client:
    result = await client.evaluate_tool_call(
        agent_id="...",
        tool_name="search",
        tool_input={"query": "test"},
        enforce=True,
    )
```

---

## API Reference

### `AgentGuard.configure()`
Configure the SDK globally. Must be called before using any other methods.

### `@AgentGuard.tool`
Decorator to protect any sync or async function. Evaluates every call before execution.

### `AgentGuard.protect_langchain_tools(tools)`
Wrap a list of LangChain `BaseTool` instances. Returns the same list with protection added.

### `AgentGuard.callback_handler()`
Return a LangChain `BaseCallbackHandler` that intercepts all tool calls.

### `AgentGuard.async_callback_handler()`
Return an async LangChain callback handler for use with `executor.arun()`.

### `AgentGuard.protect_crewai_tools(tools)`
Wrap a list of CrewAI `BaseTool` instances. Returns the same list with protection added.

### `@AgentGuard.crewai_tool`
Decorator for CrewAI tool functions. Compatible with `@crewai.tool`.

### `AgentGuard.inspect_memory(content, source, raise_on_threat)`
Inspect memory/RAG content for poisoning or injection. Returns `MemoryInspectionResult`.

### `AgentGuard.check_kill_switch()`
Return current `KillSwitchStatus` (active, reason, actor).

### `AgentGuard.assert_kill_switch_inactive()`
Raise `AgentGuardKillSwitchError` if kill switch is active.

---

## Running the Tests

```bash
cd packages/sdk-python
pip install -e ".[dev]"
pytest tests/ -v
```

---

## License

Apache 2.0 — see [LICENSE](LICENSE).

---

## Links

- [AgentGuard Documentation](https://docs.agentguard.io)
- [Dashboard](http://localhost:3000)
- [API Reference](http://localhost:8000/docs)
