Skip to content

PaloAltoNetworks/airs-api-mgmt-sdk

BranchesTags

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

3 Commits
.ci
.ci
 
 
src/airs_api_mgmt
src/airs_api_mgmt
 
 
tests
tests
 
 
.gitignore
.gitignore
 
 
.pre-commit-config.yaml
.pre-commit-config.yaml
 
 
README.md
README.md
 
 
pyproject.toml
pyproject.toml
 
 
taskfile.yml
taskfile.yml
 
 

Repository files navigation

Palo Alto Networks Prisma AIRS API Management SDK

This Python SDK provides convenient access to the Palo Alto Networks AI Runtime Security Management API for Python applications. This SDK enables you to programmatically manage AI security profiles, API keys, DLP profiles, customer applications, custom topics, OAuth tokens, and deployment profiles.

Key Features:

  • Synchronous (inline) and Asynchronous (asyncio) support - Choose the SDK version that fits your application
  • Type-safe - Fully typed with Pydantic models for all request/response objects
  • Automatic OAuth2 token management - Built-in token caching and refresh
  • Retry handling - Configurable exponential backoff for transient failures
  • Comprehensive error handling - Granular exception types for precise error handling

API Documentation

The reference API documentation for Palo Alto Networks AI Runtime Security Management API can be found at https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Installation

# Create and activate a virtual environment
python3 -m venv --prompt ${PWD##*/} .venv && source .venv/bin/activate

# Install most recent release version of airs-api-mgmt-sdk package
python3 -m pip install "pan-airs-api-mgmt-sdk"

Quick Start

Synchronous Example:

from airs_api_mgmt import MgmtClient

# Initialize client with credentials
client = MgmtClient(
    client_id="your_client_id",
    client_secret="your_client_secret"
)

# Retrieve API keys
api_keys = client.api_keys.get_all_api_keys(limit=10)
print(f"API Keys: {api_keys}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def main():
    async with MgmtClientAsync() as client:  # Uses env variables
        # Retrieve AI security profiles
        profiles = await client.ai_sec_profiles.get_all_ai_profiles(limit=10)
        print(f"AI Security Profiles: {profiles}")

asyncio.run(main())

SDK Configuration

The SDK provides two client classes:

  • MgmtClient() - Synchronous (blocking) client for standard Python applications
  • MgmtClientAsync() - Asynchronous (non-blocking) client for async/await applications

Both clients accept the following optional parameters:

  • client_id: Your OAuth2 client ID (service account email)
  • client_secret: Your OAuth2 client secret
  • base_url: Management API endpoint (default: https://api.sase.paloaltonetworks.com/aisec)
  • token_base_url: OAuth2 token endpoint (default: https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token)
  • num_retries: Maximum number of retries with exponential backoff (default: 3)

You must provide OAuth2 client credentials (client_id + client_secret) for authentication.

OAuth2 Client Credentials

There are two ways to specify your OAuth2 client credentials and configuration:

  1. Using environment variables:
# Required credentials
export PANW_CLIENT_ID="your_client_id"
export PANW_CLIENT_SECRET="your_client_secret"

# Optional configuration (override defaults)
export PANW_BASE_URL="https://api.sase.paloaltonetworks.com/aisec"
export PANW_TOKEN_BASE_URL="https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token"
  1. Specify credentials in MgmtClient() initialization:
from airs_api_mgmt import MgmtClient

client = MgmtClient(
    client_id="your_client_id",
    client_secret="your_client_secret",
    base_url="https://api.sase.paloaltonetworks.com/aisec",  # Optional
    token_base_url="https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token"  # Optional
)

Configuration Parameters

All configuration parameters:

Parameter Type Default Environment Variable Description
client_id str None PANW_CLIENT_ID OAuth2 service account client ID
client_secret str None PANW_CLIENT_SECRET OAuth2 client secret
base_url str https://api.sase.paloaltonetworks.com/aisec PANW_BASE_URL Management API endpoint
token_base_url str https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token PANW_TOKEN_BASE_URL OAuth2 token endpoint
num_retries int 3 - Max retries with exponential backoff

Example: SDK Configuration

Using Client Credentials

from airs_api_mgmt import MgmtClient

client = MgmtClient(
    client_id="your_client_id",
    client_secret="your_client_secret",
    base_url="https://api.sase.paloaltonetworks.com/aisec",
    token_base_url="https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token"
)

Using Environment Variables

from airs_api_mgmt import MgmtClient

# Set environment variables first
# Required:
# export PANW_CLIENT_ID="your_client_id"
# export PANW_CLIENT_SECRET="your_client_secret"

# Optional (override defaults):
# export PANW_BASE_URL="https://api.sase.paloaltonetworks.com/aisec"
# export PANW_TOKEN_BASE_URL="https://auth.appsvc.paloaltonetworks.com/auth/v1/oauth2/access_token"

# Initialize with defaults (uses environment variables)
client = MgmtClient()

Using Async Client

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def main():
    # Option 1: Use async context manager (recommended)
    async with MgmtClientAsync(
        client_id="your_client_id",
        client_secret="your_client_secret"
    ) as client:
        # Client automatically closes when exiting context
        api_keys = await client.api_keys.get_all_api_keys()
        print(f"API Keys: {api_keys}")

    # Option 2: Manual initialization and cleanup
    client = MgmtClientAsync()
    try:
        profiles = await client.ai_sec_profiles.get_all_ai_profiles()
        print(f"AI Profiles: {profiles}")
    finally:
        await client.close()  # Always close when done

asyncio.run(main())

Examples: Management Operations

Important: You must properly configure OAuth2 credentials or an access token before using the SDK examples.

API Keys Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Create New API Key

Create a new API key for your customer application with specified rotation settings.

Synchronous Example:

from airs_api_mgmt import MgmtClient

# Initialize client
client = MgmtClient()

# Create new API key
response = client.api_keys.create_new_api_key(
    api_key_name="my-application-key",
    cust_app="my-customer-app",
    auth_code="your_auth_code",
    created_by="user@example.com",
    cust_env="production",
    cust_cloud_provider="AWS",
    rotation_time_interval=3,
    rotation_time_unit="months",
    revoked=False,
    dp_name="my-deployment-profile",  # Optional
    cust_ai_agent_framework="langchain"  # Optional
)

print(f"API Key ID: {response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def create_api_key():
    async with MgmtClientAsync() as client:
        response = await client.api_keys.create_new_api_key(
            api_key_name="my-application-key",
            cust_app="my-customer-app",
            auth_code="your_auth_code",
            created_by="user@example.com",
            cust_env="production",
            cust_cloud_provider="AWS",
            rotation_time_interval=3,
            rotation_time_unit="months",
            revoked=False,
            dp_name="my-deployment-profile",
            cust_ai_agent_framework="langchain"
        )

        print(f"API Key ID: {response}")

asyncio.run(create_api_key())

Parameters:

  • api_key_name (str, required): Name of the API key
  • cust_app (str, required): Customer application identifier
  • auth_code (str, required): Authorization code
  • created_by (str, required): Email of user creating the key
  • cust_env (str, required): Environment (e.g., "production", "staging")
  • cust_cloud_provider (str, required): Cloud provider (e.g., "AWS", "Azure", "GCP")
  • rotation_time_interval (int, required): Rotation interval value
  • rotation_time_unit (str, required): Rotation unit ("days", "months", "years")
  • revoked (bool, optional): Revocation status (default: False)
  • dp_name (str, optional): Deployment profile name
  • cust_ai_agent_framework (str, optional): AI agent framework name

Retrieve API Keys (with pagination)

Retrieve all API keys.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve API keys with pagination
api_keys_response = client.api_keys.get_all_api_keys(offset=0, limit=25)

print(f"API Keys Response: {api_keys_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def get_all_api_keys():
    async with MgmtClientAsync() as client:
        api_keys_response = await client.api_keys.get_all_api_keys(offset=0, limit=25)

        print(f"API Keys Response: {api_keys_response}")

asyncio.run(get_all_api_keys())

Parameters:

  • offset (int, optional): Starting position for pagination (default: 0)
  • limit (int, optional): Maximum number of keys to retrieve (default: 100)

Regenerate API Key

Regenerate an existing API key with new rotation settings.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Regenerate API key
regenerate_response = client.api_keys.regenerate_api_key(
    api_key_id="your_api_key_uuid",
    rotation_time_interval=6,
    rotation_time_unit="months",
    updated_by="user@example.com"
)

print(f"New API Key Secret: {regenerate_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def regenerate_api_key():
    async with MgmtClientAsync() as client:
        regenerate_response = await client.api_keys.regenerate_api_key(
            api_key_id="your_api_key_uuid",
            rotation_time_interval=6,
            rotation_time_unit="months",
            updated_by="user@example.com"
        )

        print(f"New API Key Secret: {regenerate_response}")

asyncio.run(regenerate_api_key())

Parameters:

  • api_key_id (str, required): UUID of the API key to regenerate
  • rotation_time_interval (int, required): New rotation interval value
  • rotation_time_unit (str, required): New rotation unit ("days", "months", "years")
  • updated_by (str, optional): Email of user performing the regeneration

Delete API Key

Delete an API key by its name.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Delete API key
delete_response = client.api_keys.delete_api_key(
    api_key_name="my-application-key",
    updated_by="user@example.com"
)

print(f"Deletion response: {delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def delete_api_key():
    async with MgmtClientAsync() as client:
        delete_response = await client.api_keys.delete_api_key(
            api_key_name="my-application-key",
            updated_by="user@example.com"
        )

        print(f"Deletion response: {delete_response}")

asyncio.run(delete_api_key())

Parameters:

  • api_key_name (str, required): Name of the API key to delete
  • updated_by (str, required): Email of user performing the deletion

AI Security Profiles Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Create New AI Security Profile

Create a new AI security profile with comprehensive policy configuration.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Define AI security profile policy (using dictionary format)
policy = {
    "dlp-data-profiles": [],
    "ai-security-profiles": [
        {
            "model-type": "default",
            "model-configuration": {
                "latency": {
                    "inline-timeout-action": "block",
                    "max-inline-latency": 20
                },
                "data-protection": {
                    "data-leak-detection": {
                        "member": [
                            {"text": "Sensitive Content", "id": "", "version": "2"}
                        ],
                        "action": "block"
                    }
                },
                "app-protection": {
                    "default-url-category": {"member": ["malicious"]},
                    "url-detected-action": "allow"
                },
                "model-protection": [
                    {"name": "prompt-injection", "action": "allow"},
                    {"name": "jailbreak", "action": "block"}
                ]
            }
        }
    ]
}

# Alternatively, create policy using SDK model objects (typed approach)
from airs_api_mgmt.sdk.models import (
    AIProfileObjectPolicy,
    AiSecurityProfileObject,
    AiSecurityProfileObjectModelConfiguration,
    LatencyObject,
    DataProtectionObject,
    DataProtectionObjectDataLeakDetection,
    DataProtectionObjectDataLeakDetectionMemberInner,
    AppProtectionObject,
    AppProtectionObjectDefaultUrlCategory,
    ModelProtectionObjectInner
)

policy_typed = AIProfileObjectPolicy(
    dlp_data_profiles=[],
    ai_security_profiles=[
        AiSecurityProfileObject(
            model_type="default",
            content_type="text",
            model_configuration=AiSecurityProfileObjectModelConfiguration(
                latency=LatencyObject(
                    inline_timeout_action="block",
                    max_inline_latency=20
                ),
                data_protection=DataProtectionObject(
                    data_leak_detection=DataProtectionObjectDataLeakDetection(
                        member=[
                            DataProtectionObjectDataLeakDetectionMemberInner(
                                text="Sensitive Content",
                                id="",
                                version="2"
                            )
                        ],
                        action="block"
                    )
                ),
                app_protection=AppProtectionObject(
                    default_url_category=AppProtectionObjectDefaultUrlCategory(
                        member=["malicious"]
                    ),
                    url_detected_action="allow"
                ),
                model_protection=[
                    ModelProtectionObjectInner(name="prompt-injection", action="allow"),
                    ModelProtectionObjectInner(name="jailbreak", action="block")
                ]
            )
        )
    ]
)

# Create new AI security profile (works with both dictionary or typed policy)
create_response = client.ai_sec_profiles.create_new_ai_profile(
    profile_name="production-security-profile",
    revision=1,
    policy=policy,  # or use policy_typed for typed approach
    active=True,
    created_by="user@example.com"
)

print(f"Profile ID: {create_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def create_ai_profile():
    async with MgmtClientAsync() as client:
        # Define AI security profile policy (using dictionary format)
        policy = {
            "dlp-data-profiles": [],
            "ai-security-profiles": [
                {
                    "model-type": "default",
                    "model-configuration": {
                        "latency": {
                            "inline-timeout-action": "block",
                            "max-inline-latency": 20
                        },
                        "data-protection": {
                            "data-leak-detection": {
                                "member": [
                                    {"text": "Sensitive Content", "id": "", "version": "2"}
                                ],
                                "action": "block"
                            }
                        },
                        "app-protection": {
                            "default-url-category": {"member": ["malicious"]},
                            "url-detected-action": "allow"
                        },
                        "model-protection": [
                            {"name": "prompt-injection", "action": "allow"},
                            {"name": "jailbreak", "action": "block"}
                        ]
                    }
                }
            ]
        }

        # Alternatively, create policy using SDK model objects (typed approach)
        from airs_api_mgmt.sdk.models import (
            AIProfileObjectPolicy,
            AiSecurityProfileObject,
            AiSecurityProfileObjectModelConfiguration,
            LatencyObject,
            DataProtectionObject,
            DataProtectionObjectDataLeakDetection,
            DataProtectionObjectDataLeakDetectionMemberInner,
            AppProtectionObject,
            AppProtectionObjectDefaultUrlCategory,
            ModelProtectionObjectInner
        )

        policy_typed = AIProfileObjectPolicy(
            dlp_data_profiles=[],
            ai_security_profiles=[
                AiSecurityProfileObject(
                    model_type="default",
                    content_type="text",
                    model_configuration=AiSecurityProfileObjectModelConfiguration(
                        latency=LatencyObject(
                            inline_timeout_action="block",
                            max_inline_latency=20
                        ),
                        data_protection=DataProtectionObject(
                            data_leak_detection=DataProtectionObjectDataLeakDetection(
                                member=[
                                    DataProtectionObjectDataLeakDetectionMemberInner(
                                        text="Sensitive Content",
                                        id="",
                                        version="2"
                                    )
                                ],
                                action="block"
                            )
                        ),
                        app_protection=AppProtectionObject(
                            default_url_category=AppProtectionObjectDefaultUrlCategory(
                                member=["malicious"]
                            ),
                            url_detected_action="allow"
                        ),
                        model_protection=[
                            ModelProtectionObjectInner(name="prompt-injection", action="allow"),
                            ModelProtectionObjectInner(name="jailbreak", action="block")
                        ]
                    )
                )
            ]
        )

        # Create new AI security profile (works with both dictionary or typed policy)
        create_response = await client.ai_sec_profiles.create_new_ai_profile(
            profile_name="production-security-profile",
            revision=1,
            policy=policy,  # or use policy_typed for typed approach
            active=True,
            created_by="user@example.com"
        )

        print(f"Profile ID: {create_response}")

asyncio.run(create_ai_profile())

Parameters:

  • profile_name (str, required): Name of the AI security profile
  • revision (int, required): Revision number
  • policy (dict | AIProfileObjectPolicy, required): Policy configuration object (can be dict or typed model object)
  • profile_id (str, optional): Custom profile UUID
  • active (bool, optional): Active status
  • created_by (str, optional): Email of user creating the profile
  • updated_by (str, optional): Email of user updating the profile
  • last_modified_ts (datetime, optional): Last modification timestamp

Retrieve AI Security Profiles (with pagination)

Retrieve all AI security profiles with pagination support.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve AI security profiles
profiles_response = client.ai_sec_profiles.get_all_ai_profiles(offset=0, limit=25)

print(f"AI Security Profiles Response: {profiles_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def get_all_ai_profiles():
    async with MgmtClientAsync() as client:
        profiles_response = await client.ai_sec_profiles.get_all_ai_profiles(offset=0, limit=25)

        print(f"AI Security Profiles Response: {profiles_response}")

asyncio.run(get_all_ai_profiles())

Parameters:

  • offset (int, optional): Starting position for pagination (default: 0)
  • limit (int, optional): Maximum number of profiles to retrieve (default: 100)

Update AI Security Profile

Update an existing AI security profile by its ID.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Updated policy
updated_policy = {
    "dlp-data-profiles": [],
    "ai-security-profiles": [
        {
            "model-type": "default",
            "model-configuration": {
                "latency": {
                    "inline-timeout-action": "allow",
                    "max-inline-latency": 30
                },
                "data-protection": {
                    "data-leak-detection": {
                        "member": [
                            {"text": "PII Data", "id": "", "version": "2"}
                        ],
                        "action": "block"
                    }
                }
            }
        }
    ]
}

# Update AI security profile
update_response = client.ai_sec_profiles.update_ai_profile(
    profile_id="your_profile_uuid",
    profile_name="production-security-profile-v2",
    revision=2,
    policy=updated_policy,
    active=True,
    updated_by="user@example.com"
)

print(f"Updated Profile: {update_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def update_ai_profile():
    async with MgmtClientAsync() as client:
        updated_policy = {
            "dlp-data-profiles": [],
            "ai-security-profiles": [
                {
                    "model-type": "default",
                    "model-configuration": {
                        "latency": {
                            "inline-timeout-action": "allow",
                            "max-inline-latency": 30
                        },
                        "data-protection": {
                            "data-leak-detection": {
                                "member": [
                                    {"text": "PII Data", "id": "", "version": "2"}
                                ],
                                "action": "block"
                            }
                        }
                    }
                }
            ]
        }

        update_response = await client.ai_sec_profiles.update_ai_profile(
            profile_id="your_profile_uuid",
            profile_name="production-security-profile-v2",
            revision=2,
            policy=updated_policy,
            active=True,
            updated_by="user@example.com"
        )

        print(f"Updated Profile: {update_response.profile_name}")
        print(f"New Revision: {update_response.revision}")

asyncio.run(update_ai_profile())

Parameters:

  • profile_id (str, required): UUID of the profile to update
  • profile_name (str, required): Updated profile name
  • revision (int, required): New revision number
  • policy (dict, required): Updated policy configuration
  • active (bool, optional): Active status
  • created_by (str, optional): Original creator email
  • updated_by (str, optional): Email of user updating the profile
  • last_modified_ts (datetime, optional): Last modification timestamp

Delete AI Security Profile

Delete an AI security profile by its ID.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Delete AI security profile
delete_response = client.ai_sec_profiles.delete_ai_profile(
    profile_id="your_profile_uuid"
)

print(f"Deletion response: {delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def delete_ai_profile():
    async with MgmtClientAsync() as client:
        delete_response = await client.ai_sec_profiles.delete_ai_profile(
            profile_id="your_profile_uuid"
        )

        print(f"Deletion response: {delete_response}")

asyncio.run(delete_ai_profile())

Parameters:

  • profile_id (str, required): UUID of the profile to delete

Force Delete AI Security Profile

Force delete an AI security profile, bypassing validation checks.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Force delete AI security profile
force_delete_response = client.ai_sec_profiles.force_delete_ai_profile(
    profile_id="your_profile_uuid",
    updated_by="user@example.com"
)

print(f"Force deletion response: {force_delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def force_delete_ai_profile():
    async with MgmtClientAsync() as client:
        force_delete_response = await client.ai_sec_profiles.force_delete_ai_profile(
            profile_id="your_profile_uuid",
            updated_by="user@example.com"
        )

        print(f"Force deletion response: {force_delete_response}")

asyncio.run(force_delete_ai_profile())

Parameters:

  • profile_id (str, required): UUID of the profile to force delete
  • updated_by (str, required): Email of user performing the deletion

Custom Topics Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Create New Custom Topic

Create a new custom topic for data classification.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Create custom topic
create_response = client.custom_topics.create_new_custom_topic(
    topic_name="financial-data",
    description="Detection of financial and banking information",
    examples=[
        "Credit card numbers",
        "Bank account details",
        "Social security numbers",
        "Tax identification numbers"
    ],
    revision=1,
    created_by="user@example.com",
    active=True
)

print(f"Topic ID: {create_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def create_custom_topic():
    async with MgmtClientAsync() as client:
        create_response = await client.custom_topics.create_new_custom_topic(
            topic_name="financial-data",
            description="Detection of financial and banking information",
            examples=[
                "Credit card numbers",
                "Bank account details",
                "Social security numbers",
                "Tax identification numbers"
            ],
            revision=1,
            created_by="user@example.com",
            active=True
        )

        print(f"Topic ID: {create_response}")

asyncio.run(create_custom_topic())

Parameters:

  • topic_name (str, required): Name of the custom topic
  • description (str, required): Detailed explanation of the topic
  • examples (list[str], required): List of example usages
  • revision (int, required): Revision number
  • topic_id (str, optional): Custom topic UUID
  • active (bool, optional): Active status
  • created_by (str, optional): Email of user creating the topic
  • updated_by (str, optional): Email of user updating the topic
  • last_modified_ts (datetime, optional): Last modification timestamp
  • created_ts (datetime, optional): Creation timestamp

Retrieve All Custom Topics (with pagination)

Retrieve all custom topics with pagination support.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve all custom topics
topics_response = client.custom_topics.get_all_custom_topics(
    offset=0,
    limit=50
)

print(f"Custom Topics Response: {topics_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def retrieve_custom_topics():
    async with MgmtClientAsync() as client:
        topics_response = await client.custom_topics.get_all_custom_topics(
            offset=0,
            limit=50
        )

        print(f"Custom Topics Response: {topics_response}")

asyncio.run(retrieve_custom_topics())

Parameters:

  • offset (int, optional): Starting position for pagination (default: 0)
  • limit (int, optional): Maximum number of topics to retrieve (default: 100)

Modify Custom Topic Details

Modify an existing custom topic with updated details.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Modify custom topic
modify_response = client.custom_topics.modify_custom_topic_details(
    topic_id="your_topic_uuid",
    topic_name="financial-data-updated",
    description="Updated detection of financial and personal information",
    examples=[
        "Credit card numbers",
        "Bank account details",
        "Social security numbers",
        "Tax identification numbers",
        "IBAN numbers"
    ],
    revision=2,
    updated_by="user@example.com",
    active=True
)

print(f"Modified Topic: {modify_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def modify_custom_topic():
    async with MgmtClientAsync() as client:
        modify_response = await client.custom_topics.modify_custom_topic_details(
            topic_id="your_topic_uuid",
            topic_name="financial-data-updated",
            description="Updated detection of financial and personal information",
            examples=[
                "Credit card numbers",
                "Bank account details",
                "Social security numbers",
                "Tax identification numbers",
                "IBAN numbers"
            ],
            revision=2,
            updated_by="user@example.com",
            active=True
        )

        print(f"Modified Topic: {modify_response}")

asyncio.run(modify_custom_topic())

Parameters:

  • topic_id (str, required): UUID of the topic to modify
  • topic_name (str, required): Updated topic name
  • description (str, required): Updated description
  • examples (list[str], required): Updated list of examples
  • revision (int, required): New revision number
  • active (bool, optional): Active status
  • created_by (str, optional): Original creator email
  • updated_by (str, optional): Email of user modifying the topic
  • last_modified_ts (datetime, optional): Last modification timestamp
  • created_ts (datetime, optional): Creation timestamp

Delete Custom Topic

Delete a custom topic by its ID.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Delete custom topic
delete_response = client.custom_topics.delete_custom_topic(
    topic_id="your_topic_uuid"
)

print(f"Deletion response: {delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def delete_custom_topic():
    async with MgmtClientAsync() as client:
        delete_response = await client.custom_topics.delete_custom_topic(
            topic_id="your_topic_uuid"
        )

        print(f"Deletion response: {delete_response}")

asyncio.run(delete_custom_topic())

Parameters:

  • topic_id (str, required): UUID of the topic to delete

Force Delete Custom Topic

Force delete a custom topic, bypassing validation checks.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Force delete custom topic
force_delete_response = client.custom_topics.force_delete_custom_topic(
    topic_id="your_topic_uuid",
    updated_by="user@example.com"
)

print(f"Force deletion response: {force_delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def force_delete_custom_topic():
    async with MgmtClientAsync() as client:
        force_delete_response = await client.custom_topics.force_delete_custom_topic(
            topic_id="your_topic_uuid",
            updated_by="user@example.com"
        )

        print(f"Force deletion response: {force_delete_response}")

asyncio.run(force_delete_custom_topic())

Parameters:

  • topic_id (str, required): UUID of the topic to force delete
  • updated_by (str, required): Email of user performing the deletion

Customer Applications Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Retrieve All Customer Applications (with pagination)

Retrieve all customer applications with pagination support.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve all customer applications
apps_response = client.customer_apps.get_all_customer_apps(
    offset=0,
    limit=25
)

print(f"Customer Applications Response: {apps_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def retrieve_customer_apps():
    async with MgmtClientAsync() as client:
        apps_response = await client.customer_apps.get_all_customer_apps(
            offset=0,
            limit=25
        )

        print(f"Customer Applications Response: {apps_response}")

asyncio.run(retrieve_customer_apps())

Parameters:

  • offset (int, optional): Starting position for pagination (default: 0)
  • limit (int, optional): Maximum number of apps to retrieve (default: 100)

Update Customer Application

Update a customer application with new settings.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Update customer application
update_response = client.customer_apps.update_customer_app(
    customer_app_id="your_app_uuid",
    app_name="my-updated-application",
    cloud_provider="AWS",
    environment="production",
    model_name="gpt-4",  # Optional
    status="completed",  # Optional
    updated_by="user@example.com",  # Optional
    ai_agent_framework="langchain"  # Optional
)

print(f"Updated App: {update_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def update_customer_app():
    async with MgmtClientAsync() as client:
        update_response = await client.customer_apps.update_customer_app(
            customer_app_id="your_app_uuid",
            app_name="my-updated-application",
            cloud_provider="AWS",
            environment="production",
            model_name="gpt-4",
            status="completed",
            updated_by="user@example.com",
            ai_agent_framework="langchain"
        )

        print(f"Updated App: {update_response}")
asyncio.run(update_customer_app())

Parameters:

  • customer_app_id (str, required): UUID of the customer application
  • app_name (str, required): Updated application name
  • cloud_provider (str, required): Cloud provider ("AWS", "Azure", "GCP")
  • environment (str, required): Environment ("production", "staging", "development")
  • model_name (str, optional): AI model name
  • status (str, optional): Application status ("completed", "pending")
  • updated_by (str, optional): Email of user updating the app
  • ai_agent_framework (str, optional): AI agent framework name

Delete Customer Application

Delete a customer application by its name.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Delete customer application
delete_response = client.customer_apps.delete_customer_app(
    app_name="my-application",
    updated_by="user@example.com"
)

print(f"Deletion response: {delete_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def delete_customer_app():
    async with MgmtClientAsync() as client:
        delete_response = await client.customer_apps.delete_customer_app(
            app_name="my-application",
            updated_by="user@example.com"
        )

        print(f"Deletion response: {delete_response}")

asyncio.run(delete_customer_app())

Parameters:

  • app_name (str, required): Name of the application to delete
  • updated_by (str, required): Email of user performing the deletion

DLP Profiles Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Retrieve All DLP Profiles

Retrieve all DLP (Data Loss Prevention) profiles.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve all DLP profiles
dlp_profiles = client.dlp_profiles.get_all_dlp_profiles()

print(f"DLP Profiles: {dlp_profiles}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def retrieve_dlp_profiles():
    async with MgmtClientAsync() as client:
        # Retrieve all DLP profiles
        dlp_profiles = await client.dlp_profiles.get_all_dlp_profiles()

        print(f"DLP Profiles: {dlp_profiles}")

asyncio.run(retrieve_dlp_profiles())

Parameters: None

Deployment Profiles Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Retrieve All Deployment Profiles

Retrieve all deployment profiles.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve all deployment profiles
deployment_profiles = client.deployment_profiles.get_all_deployment_profiles()

print(f"Deployment Profiles: {deployment_profiles}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def retrieve_deployment_profiles():
    async with MgmtClientAsync() as client:
        # Retrieve all deployment profiles
        deployment_profiles = await client.deployment_profiles.get_all_deployment_profiles()

        print(f"Deployment Profiles: {deployment_profiles}")

asyncio.run(retrieve_deployment_profiles())

Parameters:

  • unactivated (bool, optional): Get only unactivated deployment profiles (default: None)

Retrieve Unactivated Deployment Profiles

Retrieve only unactivated deployment profiles (includes available and previously activated profiles without associated apps or API keys).

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Retrieve only unactivated deployment profiles
unactivated_profiles = client.deployment_profiles.get_all_deployment_profiles(
    unactivated=True
)

print(f"Unactivated Deployment Profiles: {unactivated_profiles}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def retrieve_unactivated_profiles():
    async with MgmtClientAsync() as client:
        # Retrieve only unactivated deployment profiles
        unactivated_profiles = await client.deployment_profiles.get_all_deployment_profiles(
            unactivated=True
        )

        print(f"Unactivated Deployment Profiles: {unactivated_profiles}")

asyncio.run(retrieve_unactivated_profiles())

Parameters:

  • unactivated (bool, required): Set to True to retrieve only unactivated profiles

OAuth Token Management

API Reference: https://pan.dev/prisma-airs/api/airuntimesecurity/prismaairsmanagementapi/

Generate OAuth2 Token

Generate an OAuth2 access token for an Apigee application.

Synchronous Example:

from airs_api_mgmt import MgmtClient

client = MgmtClient()

# Generate OAuth2 token for an Apigee application
oauth_response = client.oauth.get_oauth_token(
    client_id="your_apigee_client_id",
    customer_app="my-customer-app",
    token_ttl_interval=24,
    token_ttl_unit="hours"
)

print(f"OAuth Token: {oauth_response}")

Asynchronous Example:

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def generate_oauth_token():
    async with MgmtClientAsync() as client:
        # Generate OAuth2 token for an Apigee application
        oauth_response = await client.oauth.get_oauth_token(
            client_id="your_apigee_client_id",
            customer_app="my-customer-app",
            token_ttl_interval=24,
            token_ttl_unit="hours"
        )

        print(f"OAuth Token: {oauth_response}")

asyncio.run(generate_oauth_token())

Parameters:

  • client_id (str, required): Client ID for the OAuth application
  • customer_app (str, required): Customer application name
  • token_ttl_interval (int, required): Time-to-live interval for the token
  • token_ttl_unit (str, required): Time unit ("seconds", "minutes", "hours", "days")

SDK Features Deep Dive

1. Dual Sync/Async Implementation

The SDK provides both synchronous and asynchronous implementations:

  • Synchronous (MgmtClient): Traditional blocking I/O for standard Python applications
  • Asynchronous (MgmtClientAsync): Non-blocking I/O for high-performance async applications

Both implementations share the same API surface, making it easy to switch between them.

2. Automatic OAuth2 Token Management

The SDK handles OAuth2 authentication automatically:

  • Token Acquisition: Automatically obtains access tokens using client credentials
  • Token Caching: Caches tokens in memory to avoid unnecessary token requests
  • Token Refresh: Automatically refreshes tokens before expiration
  • Thread-Safe: Token management is thread-safe for concurrent requests
# Token is automatically managed - no manual intervention needed
client = MgmtClient(client_id="...", client_secret="...")
# Token acquired on first API call and refreshed as needed

3. Exponential Backoff Retry Logic

Automatic retry with exponential backoff for transient failures:

  • Default Retries: 3 attempts (configurable via num_retries parameter)
  • Backoff Strategy: Exponential backoff (1s → 2s → 4s → 8s → ...)
  • Retryable Errors: Automatically retries on network errors and 5xx server errors
  • Non-Retryable Errors: Fails fast on 4xx client errors (bad requests, auth failures)
# Configure custom retry behavior
client = MgmtClient(
    client_id="...",
    client_secret="...",
    num_retries=5  # Retry up to 5 times
)

4. Comprehensive Error Handling

Granular exception hierarchy for precise error handling:

  • MgmtSdkClientError: 4xx errors (bad requests, auth failures, not found)
  • MgmtSdkServerError: 5xx errors (server failures, timeouts)
  • PayloadConfigurationError: Invalid request payloads
  • MissingVariableError: Missing required configuration
  • MgmtSdkAPIError: General API errors
  • MgmtSdkBaseError: Unexpected errors

All HTTP error exceptions include:

  • status_code: HTTP status code
  • response_body: Raw response body for debugging

5. Type Safety with Pydantic Models

All request and response objects are validated using Pydantic v2:

  • Automatic Validation: Input/output data is automatically validated
  • Type Hints: Full type hint support for IDE autocomplete
  • Serialization: Automatic JSON serialization/deserialization
  • Documentation: Self-documenting models with field descriptions
from airs_api_mgmt.sdk.models import PaginatedAPIKeyObject

# Type-safe responses
response: PaginatedAPIKeyObject = client.api_keys.get_all_api_keys()
print(f"Response: {response}")

6. Pagination Support

Built-in pagination for all list endpoints:

  • Offset-based: Use offset and limit parameters
  • Consistent API: Same pagination pattern across all resources
  • Total Count: Responses include total count for progress tracking
# Retrieve first 50 items
page1 = client.api_keys.get_all_api_keys(offset=0, limit=50)

# Retrieve next 50 items
page2 = client.api_keys.get_all_api_keys(offset=50, limit=50)

print(f"Page 1: {page1}")
print(f"Page 2: {page2}")

7. Structured Logging

Built-in logging for debugging and monitoring:

  • Contextual Logs: Each operation logs with event context
  • Configurable: Uses standard Python logging (configure via logging module)
  • Request/Response: Logs important API interactions
  • Error Details: Detailed error logging with stack traces
import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("airs_api_mgmt")
logger.setLevel(logging.DEBUG)

client = MgmtClient()  # Now see detailed logs

8. Environment Variable Support

Flexible configuration via environment variables:

  • PANW_CLIENT_ID: OAuth2 client ID
  • PANW_CLIENT_SECRET: OAuth2 client secret
  • PANW_BASE_URL: Custom API endpoint URL
  • PANW_TOKEN_BASE_URL: Custom OAuth2 token endpoint URL
export PANW_CLIENT_ID="your_client_id"
export PANW_CLIENT_SECRET="your_client_secret"
# Automatically uses environment variables
client = MgmtClient()

9. OpenAPI-Generated Models

All models are auto-generated from OpenAPI 3.0 specification:

  • Always Current: Models stay in sync with API specification
  • Comprehensive: Complete coverage of all API endpoints
  • Validated: OpenAPI contract ensures correctness

10. Context Manager Support

Async client supports Python context managers for automatic cleanup:

async with MgmtClientAsync() as client:
    # Client automatically closes connections on exit
    profiles = await client.ai_sec_profiles.get_all_ai_profiles()
    # Cleanup happens automatically

Available Resources

The SDK provides access to the following resources through the MgmtClient:

Resource Access Method Description
API Keys client.api_keys Create, retrieve, regenerate, and delete API keys
AI Security Profiles client.ai_sec_profiles Manage AI security profiles and policies
Custom Topics client.custom_topics Create and manage custom security topics
Customer Applications client.customer_apps Manage customer applications
DLP Profiles client.dlp_profiles Retrieve DLP (Data Loss Prevention) profiles
Deployment Profiles client.deployment_profiles Retrieve deployment configurations
OAuth Tokens client.oauth Generate OAuth2 tokens for applications

All resources provide:

  • Automatic OAuth2 token management with refresh
  • Exponential backoff retry logic (1s, 2s, 4s, 8s...)
  • Type-safe request/response models
  • Comprehensive error handling

API Coverage

API Keys Management

Operation Method API Endpoint
Create API Key create_new_api_key() POST /v1/mgmt/api-key
Retrieve API Keys get_all_api_keys() GET /v1/mgmt/api-key
Regenerate API Key regenerate_api_key() POST /v1/mgmt/api-key/{apiKeyId}/regenerate
Delete API Key delete_api_key() DELETE /v1/mgmt/api-key

AI Security Profiles Management

Operation Method API Endpoint
Create AI Profile create_new_ai_profile() POST /v1/mgmt/ai-sec-profile
Retrieve AI Profiles get_all_ai_profiles() GET /v1/mgmt/ai-sec-profile
Update AI Profile update_ai_profile() PUT /v1/mgmt/ai-sec-profile/{profileId}
Delete AI Profile delete_ai_profile() DELETE /v1/mgmt/ai-sec-profile/{profileId}
Force Delete AI Profile force_delete_ai_profile() DELETE /v1/mgmt/ai-sec-profile/{profileId}/force

Custom Topics Management

Operation Method API Endpoint
Create Custom Topic create_new_custom_topic() POST /v1/mgmt/custom-topic
Retrieve Custom Topics get_all_custom_topics() GET /v1/mgmt/custom-topic
Modify Custom Topic modify_custom_topic_details() PUT /v1/mgmt/custom-topic/{topicId}
Delete Custom Topic delete_custom_topic() DELETE /v1/mgmt/custom-topic/{topicId}
Force Delete Custom Topic force_delete_custom_topic() DELETE /v1/mgmt/custom-topic/{topicId}/force

Customer Applications Management

Operation Method API Endpoint
Retrieve Customer Apps get_all_customer_apps() GET /v1/mgmt/customer-app
Update Customer App update_customer_app() PUT /v1/mgmt/customer-app/{customerAppId}
Delete Customer App delete_customer_app() DELETE /v1/mgmt/customer-app

DLP Profiles Management

Operation Method API Endpoint
Retrieve DLP Profiles get_all_dlp_profiles() GET /v1/mgmt/dlp-profile

Deployment Profiles Management

Operation Method API Endpoint
Retrieve Deployment Profiles get_all_deployment_profiles() GET /v1/mgmt/deployment-profile

OAuth Token Management

Operation Method API Endpoint
Generate OAuth Token get_oauth_token() POST /v1/oauth/token

Async SDK Usage

The SDK provides full async support through the MgmtClientAsync class. All resource methods are async-compatible and use the same API as the synchronous version.

Async Configuration

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def main():
    # Initialize async client
    async with MgmtClientAsync(
        client_id="your_client_id",
        client_secret="your_client_secret"
    ) as client:
        # Perform async operations
        api_keys = await client.api_keys.get_all_api_keys(offset=0, limit=25)
        print(f"API Keys: {api_keys}")

# Run async code
asyncio.run(main())

Async Examples

Create AI Security Profile (Async)

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def create_profile():
    async with MgmtClientAsync() as client:
        policy = {
            "dlp-data-profiles": [],
            "ai-security-profiles": [
                {
                    "model-type": "default",
                    "model-configuration": {
                        "latency": {
                            "inline-timeout-action": "block",
                            "max-inline-latency": 20
                        }
                    }
                }
            ]
        }

        profile = await client.ai_sec_profiles.create_new_ai_profile(
            profile_name="async-profile",
            revision=1,
            policy=policy,
            active=True
        )
        print(f"Created profile: {profile.profile_id}")

asyncio.run(create_profile())

Retrieve Multiple Resources in Parallel (Async)

import asyncio
from airs_api_mgmt import MgmtClientAsync

async def fetch_all_resources():
    async with MgmtClientAsync() as client:
        # Fetch multiple resources concurrently
        results = await asyncio.gather(
            client.api_keys.get_all_api_keys(),
            client.ai_sec_profiles.get_all_ai_profiles(),
            client.custom_topics.get_all_custom_topics(),
            client.dlp_profiles.get_all_dlp_profiles()
        )

        api_keys, ai_profiles, topics, dlp_profiles = results

        print(f"API Keys: {api_keys}")
        print(f"AI Profiles: {ai_profiles}")
        print(f"Custom Topics: {topics}")
        print(f"DLP Profiles: {dlp_profiles}")

asyncio.run(fetch_all_resources())

Error Handling (Async)

import asyncio
from airs_api_mgmt import MgmtClientAsync
from airs_api_mgmt.exceptions import MgmtSdkClientError, MgmtSdkServerError

async def safe_create_api_key():
    async with MgmtClientAsync() as client:
        try:
            api_key = await client.api_keys.create_new_api_key(
                api_key_name="async-key",
                cust_app="my-app",
                auth_code="auth123",
                created_by="user@example.com",
                cust_env="production",
                cust_cloud_provider="AWS",
                rotation_time_interval=3,
                rotation_time_unit="months"
            )
            print(f"API Key created: {api_key.api_key_id}")
        except MgmtSdkClientError as e:
            print(f"Client error: {e.status_code} - {e}")
        except MgmtSdkServerError as e:
            print(f"Server error: {e.status_code} - {e}")

asyncio.run(safe_create_api_key())

Model Imports

The SDK uses Pydantic models for all request and response objects. You can import models directly from the SDK for type hints and validation.

Basic Model Import

from airs_api_mgmt.sdk.models import (
    AIProfileObject,
    APIKeyObject,
    CustomTopicObject,
    CustomerAppObject,
    DLPDataProfileObject,
    DeploymentProfilesObject,
    Oauth2TokenObject
)

# Use models for type hints
def process_api_key(api_key: APIKeyObject) -> None:
    print(f"Processing API Key: {api_key.api_key_name}")
    print(f"ID: {api_key.api_key_id}")
    print(f"Created: {api_key.created_ts}")

Paginated Response Models

from airs_api_mgmt.sdk.models import (
    PaginatedAIProfileObject,
    PaginatedAPIKeyObject,
    PaginatedCustomTopicObject,
    PaginatedCustomerAppObject
)

# Type-safe pagination handling
def display_paginated_profiles(response: PaginatedAIProfileObject) -> None:
    print(f"Paginated Profiles Response: {response}")

Policy and Configuration Models

from airs_api_mgmt.sdk.models import (
    AiSecurityProfileObjectModelConfiguration,
    DataProtectionObject,
    AppProtectionObject,
    LatencyObject
)

# Build policy with type-safe models
def create_policy_config() -> dict:
    return {
        "dlp-data-profiles": [],
        "ai-security-profiles": [
            {
                "model-type": "default",
                "model-configuration": {
                    "latency": {
                        "inline-timeout-action": "block",
                        "max-inline-latency": 20
                    },
                    "data-protection": {
                        "data-leak-detection": {
                            "member": [{"text": "PII", "id": "", "version": "2"}],
                            "action": "block"
                        }
                    }
                }
            }
        ]
    }

Array and Collection Models

from airs_api_mgmt.sdk.models import (
    ArrayDlpProfileObject,
    TopicArray
)

# Handle array responses
def process_dlp_profiles(profiles: ArrayDlpProfileObject) -> None:
    if profiles.dlp_profiles:
        for profile in profiles.dlp_profiles:
            print(f"DLP Profile: {profile.profile_name}")
    else:
        print("No DLP profiles found")

Complete Type-Safe Example

from airs_api_mgmt import MgmtClient
from airs_api_mgmt.sdk.models import (
    APIKeyObject,
    PaginatedAPIKeyObject,
    AIProfileObject,
    CustomTopicObject
)

def manage_resources() -> None:
    client = MgmtClient()

    # Type-safe API key creation
    new_key: APIKeyObject = client.api_keys.create_new_api_key(
        api_key_name="typed-key",
        cust_app="my-app",
        auth_code="auth123",
        created_by="user@example.com",
        cust_env="production",
        cust_cloud_provider="AWS",
        rotation_time_interval=3,
        rotation_time_unit="months"
    )

    # Type-safe pagination
    keys_response: PaginatedAPIKeyObject = client.api_keys.get_all_api_keys()
    print(f"API Keys Response: {keys_response}")

Error Handling & Exceptions

When the client is unable to fetch the expected response from the API server, a subclass of airs_api_mgmt.exceptions.MgmtSdkException is raised.

There are five types of exceptions defined in airs_api_mgmt/exceptions.py:

  • MgmtSdkServerError: Server-side errors (5xx status codes)

    • Includes status_code and response_body attributes

  • MgmtSdkClientError: Client-side errors (4xx status codes - invalid requests, authentication failures, resource not found)

    • Includes status_code and response_body attributes

  • PayloadConfigurationError: Invalid request payload or configuration errors

    • Raised before API calls when payload validation fails

  • MgmtSdkAPIError: General API errors that don't fit other categories

  • MgmtSdkBaseError: Base exception class for unexpected errors

  • MissingVariableError: Configuration errors (e.g., missing required credentials)

  • MgmtSdkException: Base exception class for all SDK errors

Example Error Handling

from airs_api_mgmt import MgmtClient
from airs_api_mgmt.exceptions import (
    MgmtSdkServerError,
    MgmtSdkClientError,
    PayloadConfigurationError,
    MissingVariableError,
    MgmtSdkAPIError,
    MgmtSdkBaseError,
)

client = MgmtClient()

try:
    # Define policy
    policy = {
        "dlp-data-profiles": [],
        "ai-security-profiles": [
            {
                "model-type": "default",
                "model-configuration": {
                    "latency": {
                        "inline-timeout-action": "block",
                        "max-inline-latency": 20
                    }
                }
            }
        ]
    }

    profile = client.ai_sec_profiles.create_new_ai_profile(
        profile_name="test-profile",
        revision=1,
        policy=policy,
        active=True,
        created_by="user@example.com"
    )
    print(f"Profile created: {profile.profile_id}")

except PayloadConfigurationError as e:
    print(f"Invalid payload: {e}")
    # Handle validation errors (e.g., missing required fields, invalid formats)

except MgmtSdkClientError as e:
    print(f"Client error (4xx): {e}")
    print(f"Status code: {e.status_code}")
    print(f"Response body: {e.response_body}")
    # Handle authentication failures, invalid requests, resource not found

except MgmtSdkServerError as e:
    print(f"Server error (5xx): {e}")
    print(f"Status code: {e.status_code}")
    print(f"Response body: {e.response_body}")
    # Handle server-side errors, retry logic

except MgmtSdkAPIError as e:
    print(f"API error: {e}")
    # Handle general API errors

except MissingVariableError as e:
    print(f"Configuration error: {e}")
    # Handle missing credentials or configuration

except MgmtSdkBaseError as e:
    print(f"Unexpected SDK error: {e}")
    # Handle unexpected errors during execution

Error Handling Best Practices

  1. Catch specific exceptions first: Always catch more specific exceptions before general ones
  2. Log error details: Use status_code and response_body for debugging
  3. Implement retry logic: For MgmtSdkServerError, consider implementing retry with exponential backoff
  4. Validate payloads: Handle PayloadConfigurationError to fix invalid inputs before retrying
  5. Check credentials: Handle MissingVariableError early in your application startup

Compatibility Policy

This package generally follows SemVer v2 conventions, though certain backwards-incompatible changes may be released as minor versions:

  1. Changes that only affect static types, without breaking runtime behavior.
  2. Changes to library internals which are technically public but not intended or documented for external use.
  3. Changes that we do not expect to impact the vast majority of users in practice.

We take backwards-compatibility seriously and work hard to ensure you can rely on a smooth upgrade experience. The major version number will be consistent with the API major version.

Python Version Support

  • Minimum: Python 3.10+
  • Tested: Python 3.10, 3.11, 3.12, 3.13, 3.14

Legal

Copyright (c) 2025, Palo Alto Networks

Licensed under the Polyform Internal Use License 1.0.0 (the "License"); you may not use this file except in compliance with the License.

You may obtain a copy of the License at:

https://polyformproject.org/licenses/internal-use/1.0.0

(or)

https://github.com/polyformproject/polyform-licenses/blob/76a278c4/PolyForm-Internal-Use-1.0.0.md

As far as the law allows, the software comes as is, without any warranty or condition, and the licensor will not be liable to you for any damages arising out of these terms or the use or nature of the software, under any kind of legal claim.

About

Prisma AIRS Runtime Management API - Python SDK

Resources

Readme

Code of conduct

Code of conduct

Contributing

Contributing

Security policy

Security policy
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages