cogniify-agentwatch 0.1.1

AgentWatch SDK — AI agent governance, audit, and policy enforcement

AgentWatch SDK

The official Python SDK for AgentWatch — AI agent governance, audit logging, and real-time policy enforcement.

Every tool call your AI agent makes is recorded, checked against your policies, and visible in the AgentWatch dashboard. If a policy blocks an action, the SDK stops execution before it happens.

---

Installation

pip install agentwatch

With framework integrations:

pip install agentwatch[langchain]   # LangChain
pip install agentwatch[openai]      # OpenAI

---

Quickstart

1. Register your agent

Log into the AgentWatch dashboard → Agents → Register Agent. Copy the agent UUID that gets generated.

2. Pick your integration style

---

Option A — Decorator (recommended)

Set up once at startup. Every decorated function is automatically logged and policy-checked before it runs.

import agentwatch

agentwatch.configure(
    agent_id="your-agent-uuid",
    human_owner="[e-mail]",
    api_url="https://your-agentwatch.com/api/v1",
)

@agentwatch.monitor(target_system="crm")
def read_customer(customer_id: str) -> dict:
    return crm.get(customer_id)

@agentwatch.monitor(target_system="email")
def send_followup(to: str, body: str):
    email.send(to=to, body=body)

If a policy blocks the action, BlockedByPolicy is raised before the function body executes.

---

Option B — Explicit client

Full control over every event. Useful when you need to inspect the policy result before deciding what to do.

from agentwatch import AgentWatch, BlockedByPolicy

with AgentWatch(
    agent_id="your-agent-uuid",
    human_owner="[e-mail]",
    api_url="https://your-agentwatch.com/api/v1",
) as sdk:

    result = sdk.record_event(
        action_type="read_crm",
        target_system="crm",
        input_summary=f"read record {customer_id}",
        output_summary=str(record),
    )

    if sdk.is_blocked(result):
        raise RuntimeError(f"Blocked: {result['violations']}")

---

Option C — LangChain

Pass the callback handler once. Every tool call in your agent loop is captured automatically.

from langchain_anthropic import ChatAnthropic
from agentwatch.integrations.langchain import AgentWatchCallbackHandler

handler = AgentWatchCallbackHandler(
    agent_id="your-agent-uuid",
    human_owner="[e-mail]",
    api_url="https://your-agentwatch.com/api/v1",
)

llm = ChatAnthropic(model="claude-sonnet-4-6", callbacks=[handler])
# Nothing else changes in your agent code.

---

Option D — OpenAI

Wrap the client once. All tool calls in model responses are recorded before your code acts on them.

import openai
from agentwatch.integrations.openai import wrap_openai

client = wrap_openai(
    openai.OpenAI(),
    agent_id="your-agent-uuid",
    human_owner="[e-mail]",
    api_url="https://your-agentwatch.com/api/v1",
)

# Use client exactly as before.
response = client.chat.completions.create(
    model="gpt-4o",
    tools=[...],
    messages=[...],
)

---

Configuration

agentwatch.configure()

Parameter	Type	Required	Default	Description
agent_id	str	Yes	—	UUID from the AgentWatch dashboard
human_owner	str	Yes	—	Email of the person responsible for this agent
api_url	str	No	http://localhost:8000/api/v1	URL of your AgentWatch backend
agent_name	str	No	None	Display name (overrides dashboard name)
silent	bool	No	False	Suppress console output

AgentWatch()

Same parameters as configure(), plus:

Parameter	Type	Default	Description
session_id	str	auto-generated UUID	Group related events into a session
raise_on_block	bool	False	Raise BlockedByPolicy automatically instead of returning a dict

@agentwatch.monitor()

Parameter	Type	Default	Description
action_type	str	function name	Override the logged action name
target_system	str	"unknown"	The system being accessed (e.g. "crm", "email")
permission_scope	list[str]	None	Permission tags to attach to the event

---

Handling blocked actions

from agentwatch import AgentWatch, BlockedByPolicy

sdk = AgentWatch(agent_id="...", human_owner="...", raise_on_block=True)

try:
    sdk.record_event("delete_record", "crm", "delete C-1001", "")
except BlockedByPolicy as e:
    print(f"Blocked by: {e.violations}")   # ['tool_denylist']

Or check manually:

sdk = AgentWatch(agent_id="...", human_owner="...")
result = sdk.record_event(...)
if sdk.is_blocked(result):
    # result["violations"] contains the list of matched policy names
    return {"error": "Action not permitted"}

---

Session management

Each AgentWatch instance holds a session ID that groups events together in the audit log. To start a new session for a new task run without creating a whole new client:

sdk = AgentWatch(agent_id="...", human_owner="...")

# First task
run_task(sdk, task="Look up account C-1001")

# New session for next task
sdk = sdk.new_session()
run_task(sdk, task="Send follow-up emails")

---

Fault tolerance

AgentWatch is designed to never be a point of failure. If the AgentWatch backend is unreachable, record_event logs a warning and returns {"policy_status": "compliant"} so your agent continues uninterrupted.

# This will never raise even if AgentWatch is down
result = sdk.record_event(...)   # returns safe default if server unreachable

---

What shows up in the dashboard

Every record_event call creates one row in the audit log with:

• Agent name and session ID
• Action type and target system
• Input and output summaries
• Policy status (compliant / flagged / blocked)
• Which policies were violated (if any)
• Timestamp

Flagged and blocked events generate alerts. Unusual patterns trigger anomaly detection with AI-generated explanations.

---

Requirements

• Python 3.9+
• httpx >= 0.27
• LangChain integration: langchain-core >= 0.1
• OpenAI integration: openai >= 1.0

---

License

MIT