argus-sdk

Python SDK for Argus — AI agent observability and security monitoring.

Argus captures every LLM call, tool call, and action your agent takes, scores them for risk in real time, and surfaces alerts for dangerous operations, PII exposure, secret leakage, and prompt injection.

Installation

pip install argus-agent-sdk

With LangChain support:

pip install "argus-agent-sdk[langchain]"

Quick Start

from argus_sdk import ArgusClient

argus = ArgusClient(
    api_key="sk_live_...",   # from your Argus dashboard
    actor_id="my-agent",     # unique ID for this agent
)

# Track an LLM call
argus.llm_call(
    input_text=prompt,
    output_text=response,
    input_tokens=512,
    output_tokens=128,
)

# Track a tool call
argus.tool_call(
    action_type="search_web",
    input_text=query,
    output_text=result,
)

# Track any other action
argus.action(
    action_type="file_write",
    payload={"path": "/tmp/output.txt"},
)

LangChain Integration

Drop ArgusLangChainHandler into any chain or agent — it automatically captures all LLM calls and tool calls:

from argus_sdk import ArgusClient, ArgusLangChainHandler
from langchain.chains import LLMChain

argus = ArgusClient(api_key="sk_live_...", actor_id="my-agent")
handler = ArgusLangChainHandler(argus)

chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
result = chain.run("...")
# LLM calls and tool calls are automatically audited

OpenAI Wrapper

Wrap your OpenAI client to audit all chat.completions.create() calls with zero changes to your application code:

from openai import OpenAI
from argus_sdk import ArgusClient, wrap_openai

client = wrap_openai(
    OpenAI(),
    ArgusClient(api_key="sk_live_...", actor_id="my-agent"),
)

# All calls below are automatically audited
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

Workflow Correlation

Group related events into a workflow chain using correlation_id:

import uuid

correlation_id = str(uuid.uuid4())

argus.llm_call(input_text="Plan the task", correlation_id=correlation_id)
argus.tool_call(action_type="read_file", correlation_id=correlation_id)
argus.tool_call(action_type="write_file", correlation_id=correlation_id)
argus.llm_call(output_text="Done", correlation_id=correlation_id)

Argus groups these into a single workflow run and computes chain-level risk (peak score, cumulative score, escalation detection).

API Reference

ArgusClient(api_key, actor_id, base_url=...)

argus.llm_call(...)

argus.tool_call(...) / argus.action(...)

All methods return the parsed JSON response dict or None on failure. Errors are printed as warnings and never raise — Argus is designed to be non-blocking in production.

Self-Hosted

Point the SDK at your own Argus deployment:

argus = ArgusClient(
    api_key="sk_live_...",
    actor_id="my-agent",
    base_url="https://argus.yourcompany.com",
)

Requirements

• Python ≥ 3.9
• requests ≥ 2.28
• langchain ≥ 0.1.0 (optional, only needed for ArgusLangChainHandler)

License

MIT