Quickstart

This guide walks through installing the ARX SDK, authenticating with the platform, executing a governed operation against CrowdStrike Falcon, and configuring a policy rule. By the end, you will have a working agent that reads detections through ARX with full audit logging and policy evaluation.

Prerequisites

Step 1: Install the SDK

Install the ARX SDK from PyPI:

pip install agentvault-sdk

The SDK requires httpx and structlog as dependencies, which are installed automatically.

Verify the installation:

python -c "from agentvault import ARXClient; print('SDK installed')"

Step 2: Get Your API Key

  1. Log in to the ARX console at https://console.arxsec.io.
  2. Navigate to Settings > API Keys.
  3. Click Create API Key.
  4. Assign the key a descriptive name (e.g., triage-agent-dev).
  5. Select the appropriate role. For this quickstart, deployer is sufficient.
  6. Copy the key immediately -- it will not be displayed again.

Step 3: Set Environment Variables

Export your API key and the ARX API endpoint:

export ARX_API_KEY="arx_sk_your_api_key_here"
export ARXSEC_API_URL="https://api.arxsec.io"

The SDK reads these environment variables automatically. You can also pass them directly to the ARXClient constructor.

Step 4: Initialize ARXClient

Create a new Python file named quickstart.py:

import asyncio
import os
from agentvault import ARXClient


async def main():
    # Initialize the client. Reads ARX_API_KEY and ARXSEC_API_URL
    # from environment variables if not provided explicitly.
    async with ARXClient() as arx:
        print(f"Connected to {arx.api_url}")


asyncio.run(main())

Run it:

python quickstart.py

If the API key is valid, you will see Connected to https://api.arxsec.io. If the key is missing or invalid, the SDK raises a ValueError with a descriptive message.

The async with pattern ensures the underlying HTTP client is properly closed when the block exits. You can also manage the lifecycle manually:

arx = ARXClient(api_key="arx_sk_...", api_url="https://api.arxsec.io")
# ... use the client ...
await arx.close()

Step 5: Create a Connector

The SDK provides typed connector factories for each supported tool. Call the factory method on ARXClient to get an SDKConnector instance:

async with ARXClient() as arx:
    cs = arx.crowdstrike()

The SDKConnector object does not hold credentials or connect to CrowdStrike directly. All operations are routed through the ARX API, which retrieves credentials from the secrets vault, evaluates policy, logs the action, and then executes the operation against CrowdStrike.

Step 6: Execute an Operation

Query the 10 most recent CrowdStrike detections:

import asyncio
from agentvault import ARXClient


async def main():
    async with ARXClient() as arx:
        cs = arx.crowdstrike()

        # Using the convenience method
        detections = await cs.get_detections(limit=10)
        print(f"Found {len(detections.get('resources', []))} detections")

        # Equivalent using the generic execute() method
        detections = await cs.execute("detections:read", {"limit": 10})
        print(f"Found {len(detections.get('resources', []))} detections")


asyncio.run(main())

Both approaches produce the same result. Convenience methods like get_detections() wrap execute() with typed parameters. The generic execute(operation, params) method works for any operation defined in the connector's operation map.

Behind the scenes, the SDK sends a POST request to /v1/sdk/execute with the connector name, operation identifier, parameters, and session context. The ARX API handles the rest.

Reading Detection Details

async with ARXClient() as arx:
    cs = arx.crowdstrike()

    # Get detection IDs
    result = await cs.get_detections(limit=5)
    detection_ids = result.get("resources", [])

    # Fetch full details
    if detection_ids:
        details = await cs.get_detection_details(ids=detection_ids)
        for d in details.get("resources", []):
            print(f"{d['detection_id']}: {d.get('max_severity_displayname', 'Unknown')}")

Step 7: Verify in the Audit Trail

Every operation executed through ARX produces an audit event. To verify that your operations were logged:

  1. Open the ARX console at https://console.arxsec.io.
  2. Navigate to Audit Trail.
  3. Filter by connector crowdstrike and the time range of your test.
  4. You should see entries for each detections:read and detections:read_details operation, including the policy verdict (PERMIT), risk score, and execution result.

You can also query the audit trail programmatically via the REST API:

curl -s -H "Authorization: Bearer $ARX_API_KEY" \
  "https://api.arxsec.io/v1/audit?connector=crowdstrike&limit=5" | python -m json.tool

Step 8: Set a Policy Rule

Create a policy rule that requires human approval for host containment operations. Host containment is classified as high risk in the CrowdStrike connector.

  1. In the ARX console, navigate to Policies > Rules.
  2. Click Create Rule.
  3. Configure the rule:
  4. Name: Require approval for host containment
  5. Connector: crowdstrike
  6. Operation: hosts:write
  7. Verdict: ESCALATE
  8. Priority: 100
  9. Save the rule.

You can also create rules via the API:

curl -X POST "https://api.arxsec.io/v1/policies/rules" \
  -H "Authorization: Bearer $ARX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Require approval for host containment",
    "connector": "crowdstrike",
    "operation": "hosts:write",
    "verdict": "ESCALATE",
    "priority": 100
  }'

With this rule in place, any agent that attempts hosts:write (which includes contain, lift containment, hide, and unhide actions) will trigger an approval request instead of executing immediately.

Step 9: Trigger an Escalation

Test the policy rule by attempting a host containment operation:

import asyncio
from agentvault import ARXClient


async def main():
    async with ARXClient() as arx:
        cs = arx.crowdstrike()

        try:
            # This will trigger an ESCALATE verdict based on the policy rule
            result = await cs.contain_host(ids=["target_host_id"])
            print("Host contained:", result)
        except Exception as e:
            # Expected: the operation requires human approval
            print(f"Escalation triggered: {e}")


asyncio.run(main())

When this code runs, ARX evaluates the hosts:write operation against the policy rule you created. Because the verdict is ESCALATE, ARX creates an approval request and sends a notification to the configured Slack channel or Microsoft Teams channel. The operation does not execute until a human reviewer approves it.

Check the Approval Requests page in the ARX console to see the pending request. The audit trail will show the operation with an ESCALATE verdict.

Next Steps