argus-agent-sdk 0.1.0

Python SDK for Argus — AI agent observability and security monitoring

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=...)

Parameter	Type	Required	Description
api_key	str	✓	Your Argus API key (sk_live_...)
actor_id	str	✓	Unique identifier for this agent
base_url	str		Argus deployment URL (default: https://app.argus.ai)

argus.llm_call(...)

Parameter	Type	Description
input_text	str	Prompt sent to the model
output_text	str	Model response
input_tokens	int	Prompt token count
output_tokens	int	Completion token count
correlation_id	str	Workflow chain ID
payload	dict	Arbitrary metadata

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

Parameter	Type	Description
action_type	str	Name of the tool or action (e.g. file_delete, search_web)
input_text	str	Input to the tool
output_text	str	Output from the tool
correlation_id	str	Workflow chain ID
payload	dict	Arbitrary metadata

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