Using Connectors¶

Connectors are the interface between your agent code and the security tools governed by ARX. This page covers the SDKConnector class, the generic execute() method, operation identifiers, risk classifications, approval flows, and practical code examples.

SDKConnector Overview¶

When you call a connector factory method on ARXClient (e.g., arx.crowdstrike()), it returns an SDKConnector instance. This object is a lightweight proxy -- it does not hold credentials or maintain a connection to the target tool. All operations are routed through the ARX API, where the governance layer handles credential injection, policy evaluation, and audit logging.

from agentvault import ARXClient

async with ARXClient() as arx:
    cs = arx.crowdstrike()   # Returns SDKConnector bound to "crowdstrike"
    splunk = arx.splunk()    # Returns SDKConnector bound to "splunk"
    jira = arx.jira()        # Returns SDKConnector bound to "jira"

Each SDKConnector is bound to a specific connector name. You can create multiple connectors from the same ARXClient instance. They share the same HTTP client, API key, and session context.

The execute() Method¶

Every connector operation is routed through a single method:

result = await connector.execute(operation: str, params: dict | None = None) -> dict

Parameters:

Parameter	Type	Description
`operation`	`str`	The operation identifier (e.g., `detections:read`, `hosts:write`). Must match an entry in the connector's operation map.
`params`	`dict \\| None`	Operation-specific parameters. Passed to the target tool API. Defaults to `{}` if not provided.

Returns: A dict containing the response from the target tool API. The structure depends on the specific tool and operation.

Raises:

PermissionDeniedError -- The operation was denied by policy (DENY verdict).
ApprovalDeniedError -- The operation was escalated and the human reviewer denied it.
httpx.HTTPStatusError -- The ARX API returned an error (authentication failure, rate limit, server error).

How execute() Works Internally¶

When you call execute(), the SDK sends a POST /v1/sdk/execute request to the ARX API with the following payload:

{
  "connector": "crowdstrike",
  "operation": "detections:read",
  "params": {"limit": 10},
  "session_context": {
    "prior_actions": [
      {"connector": "crowdstrike", "operation": "detections:read"},
      {"connector": "crowdstrike", "operation": "detections:read_details"}
    ]
  }
}

The server-side governance layer then:

Retrieves connector credentials from the secrets vault.
Evaluates the operation against the organization's policy rules.
Writes an immutable audit event.
Routes on the verdict (PERMIT / ESCALATE / DENY).
Executes the operation against the target tool API (if permitted).
Records the result in the audit trail.

Operation Identifiers and the OPERATION_MAP¶

Each connector defines an OPERATION_MAP -- a dictionary that maps operation identifiers to API endpoint definitions. An operation entry contains:

Field	Description
`method`	HTTP method (`GET`, `POST`, `PATCH`, `DELETE`)
`path`	API endpoint path on the target tool
`scope`	Required API scope/permission on the target tool
`risk`	Risk classification: `low`, `medium`, `high`, or `critical`
`description`	Human-readable description of the operation

For example, the CrowdStrike connector maps:

detections:read         → GET  /detects/queries/detects/v1          (low)
detections:read_details → POST /detects/entities/summaries/GET/v1   (low)
detections:write        → PATCH /detects/entities/detects/v2        (medium)
hosts:read              → GET  /devices/queries/devices-scroll/v1   (low)
hosts:write             → POST /devices/entities/devices-actions/v2 (high)
rtr:command_admin       → POST /real-time-response/entities/admin-command/v1 (critical)

You do not interact with the OPERATION_MAP directly. Pass the operation identifier string to execute(), and the server resolves it to the correct API call.

Risk Classifications¶

Operations are classified into four risk levels. These classifications are set per-operation in the connector definition and influence policy evaluation and risk scoring.

Level	Description	Examples
low	Read-only queries and status checks. No state modification.	`detections:read`, `hosts:read`, `rtr:session_list`
medium	Write operations that modify state with limited blast radius.	`detections:write` (status change), `incidents:write`, `iocs:write`
high	Operations with significant impact potential. Containment, session initialization, file operations.	`hosts:write` (contain/lift), `rtr:session_init`, `rtr:command_execute`
critical	Destructive or highly privileged operations. Active response commands, admin-level execution, batch operations.	`rtr:command_admin`, `rtr:command_active_responder`, `rtr:batch_*`

Organizations typically configure policy rules based on these risk levels. A common baseline:

low operations: PERMIT by default.
medium operations: PERMIT with elevated logging.
high operations: ESCALATE (require human approval).
critical operations: DENY unless explicitly permitted for specific agents.

Handling ESCALATE Verdicts¶

When a policy rule returns an ESCALATE verdict, the ARX API creates an approval request and sends notifications to the configured Slack or Microsoft Teams channels. The operation blocks until a reviewer acts on the request.

From the agent's perspective, an ESCALATE verdict that has not yet been approved raises an exception:

from agentvault import ARXClient

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

    try:
        result = await cs.execute("hosts:write", {
            "ids": ["host_abc123"],
            "action_name": "contain"
        })
        print("Host contained:", result)
    except Exception as e:
        print(f"Operation escalated or denied: {e}")
        # The approval request ID is available in the ARX console
        # and via the audit trail API.

To build agents that handle escalations gracefully, check the exception type and implement retry logic or alternative workflows:

try:
    result = await cs.contain_host(ids=["host_abc123"])
except ApprovalDeniedError:
    # Human reviewer denied. Log and escalate through alternate channel.
    await jira.create_issue(
        project="SEC",
        summary="Manual containment required: host_abc123",
        issue_type="Task",
        priority="High",
    )
except PermissionDeniedError:
    # Policy hard-denies this operation for this agent.
    log.error("Agent lacks permission for host containment")

Convenience Methods¶

The SDK provides typed convenience methods for common operations on frequently used connectors. These wrap execute() with named parameters.

CrowdStrike¶

cs = arx.crowdstrike()

# Query detections with optional FQL filter
detections = await cs.get_detections(filter="status:'new'", limit=50)

# Get full detection details by ID
details = await cs.get_detection_details(ids=["ldt_abc123", "ldt_def456"])

# Update detection status
await cs.update_detection(ids=["ldt_abc123"], status="in_progress")

# Query hosts
hosts = await cs.get_hosts(filter="platform_name:'Windows'", limit=100)

# Contain a host (high risk -- may trigger ESCALATE)
await cs.contain_host(ids=["host_abc123"])

# Query incidents
incidents = await cs.get_incidents(filter="fine_score:>=75", limit=20)

Splunk¶

splunk = arx.splunk()

# Run a search query
search = await splunk.search(
    query="search index=main sourcetype=syslog error",
    earliest="-4h",
    latest="now",
)

# Get search results by SID
results = await splunk.get_search_results(sid=search["sid"])

Jira¶

jira = arx.jira()

# Search issues with JQL
issues = await jira.search_issues(
    jql="project = SEC AND status = Open ORDER BY priority DESC",
    max_results=25,
)

# Create an issue
new_issue = await jira.create_issue(
    project="SEC",
    summary="Investigate anomalous login pattern on host-prod-42",
    issue_type="Task",
    description="CrowdStrike detection ldt_abc123 flagged repeated failed logins.",
    priority="High",
)

# Add a comment to an existing issue
await jira.add_comment(
    issue_key="SEC-1042",
    body="Automated enrichment complete. See attached IOC report.",
)

Using the Generic execute() for Any Connector¶

For connectors without convenience methods, or when using operations not covered by convenience methods, use execute() directly:

# Okta: List users
okta = arx.okta()
users = await okta.execute("users:read", {"limit": 200})

# Wiz: Get issues
wiz = arx.wiz()
issues = await wiz.execute("issues:read", {"first": 50})

# ServiceNow: Create an incident
snow = arx.servicenow()
incident = await snow.execute("incidents:write", {
    "short_description": "Automated: Critical vulnerability detected",
    "urgency": 1,
    "impact": 1,
})

# PagerDuty: Trigger an event
pd = arx.pagerduty()
await pd.execute("events:write", {
    "routing_key": "your_integration_key",
    "event_action": "trigger",
    "payload": {
        "summary": "Critical detection on host-prod-42",
        "severity": "critical",
        "source": "arx-triage-agent",
    },
})

Pagination¶

Most read operations that return lists support limit and offset parameters. The exact parameter names depend on the target tool's API conventions:

# CrowdStrike uses limit/offset
page1 = await cs.execute("detections:read", {"limit": 100, "offset": 0})
page2 = await cs.execute("detections:read", {"limit": 100, "offset": 100})

# Jira uses maxResults/startAt
page1 = await jira.execute("issues:read", {"jql": "project=SEC", "maxResults": 50, "startAt": 0})
page2 = await jira.execute("issues:read", {"jql": "project=SEC", "maxResults": 50, "startAt": 50})

Each paginated call is an independent operation that passes through the full governance pipeline (policy evaluation, audit logging). This means that each page fetch is individually audited and policy-evaluated.

Available Connectors¶

The SDK provides factory methods for all 101 supported connectors. Common connectors include:

Method	Connector
`arx.crowdstrike()`	CrowdStrike Falcon
`arx.splunk()`	Splunk Enterprise / Cloud
`arx.jira()`	Atlassian Jira
`arx.slack()`	Slack
`arx.wiz()`	Wiz Cloud Security
`arx.servicenow()`	ServiceNow
`arx.okta()`	Okta Identity
`arx.palo_alto()`	Palo Alto Networks
`arx.sentinel()`	Microsoft Sentinel
`arx.sentinelone()`	SentinelOne
`arx.elastic_security()`	Elastic Security
`arx.prisma_cloud()`	Prisma Cloud
`arx.defender()`	Microsoft Defender
`arx.snyk()`	Snyk

For the full list, see the ARXClient class reference or the connector catalog in the ARX console.