Add AI Access Control Example

docs/examples/ai-access-control.md

@@ -0,0 +1,100 @@
+# AI Access Control
+
+Example of implementing access control security for AI agents using PydanticAI and Permit.io
+
+Demonstrates:
+
+- [tools](../tools.md)
+- [agent dependencies](../dependencies.md)
+- [structured `result_type`](../results.md#structured-result-validation)
+- [result validation](../results.md#result-validators-functions)
+- Permit.io [AI Access Control](https://www.permit.io/ai-access-control)
+
+In this `Financial Advisor` agent example, the user can ask for financial advice, and the agent will use the `validate_financial_query` and `validate_financial_response` tools to check permissions and validate responses.
+The raw user query is transformed into a `FinancialQuery` object that includes user context (such as their permission tier).
+This structured data allows us to carry out permission checks before the AI generates a response, and the output is strictly validated with `FinancialResponse`.
+
+```mermaid
+graph TD
+    USER[User Query] -->|Request| VALIDATE[Validate Financial Query]
+    VALIDATE -->|Permission Check| PERMIT[Permit.io]
+    PERMIT -->|Decision| VALIDATE
+    VALIDATE -->|If Approved| AI[AI Processing]
+    AI -->|Generate Response| VALIDATE_RESP[Validate Financial Response]
+    VALIDATE_RESP -->|Permission Check| PERMIT
+    PERMIT -->|Decision| VALIDATE_RESP
+    VALIDATE_RESP -->|Add Disclaimer if Needed| RESPONSE[Return Structured Response]
+    RESPONSE -->|Final Response| USER
+```
+
+## Running the Example
+
+### Setup Access Control Configuration
+To run this example, you need a Permit.io API key and a running Permit authorization microservice
+
+- Get a free API key [here](https://app.permit.io) set it as an environment variable with the key `PERMIT_KEY`
+- Run a local Policy Decision Point (PDP):
+
+```bash
+docker pull permitio/pdp-v2:latest
+docker run -it \\
+    -p 7766:7000 \\
+    --env PDP_API_KEY=<YOUR_API_KEY> \\
+    --env PDP_DEBUG=True \\
+    permitio/pdp-v2:latest
+```
+
+Run the following command to set up our Permit.io financial advisor policy configuration:
+
+```bash
+python/uv-run -m pydantic_ai_examples.ai-access-control-config
+```
+
+### Run the Example
+
+With [dependencies installed and environment variables set](./index.md#usage), run the following command to run the example:
+
+```bash
+python/uv-run -m pydantic_ai_examples.ai-access-control
+```
+
+## Example code
+
+```python {title="pydantic_ai_examples/ai-access-control.py"}
+#! examples/pydantic_ai_examples/ai-access-control.py
+```
+
+## Configuration code
+
+The following script will set the following fine-grained Attribute-Based Access Control (ABAC) policy in Permit.io
+
+1. **User Tiers**:
+   - `opted_in_user`: Users who have explicitly consented to AI advice
+   - `premium_user`: Users with access to all features and advanced financial advice
+2. **Resources and Actions**:
+   - `financial_advice`: Controls who can receive investment recommendations
+     - Attributes: `is_ai_generated` (boolean)
+     - Actions: `receive`
+   - `financial_response`: Controls content and disclaimers in responses
+     - Attributes: `contains_advice` (boolean)
+     - Actions: `requires_disclaimer`
+3. **Decision Logic**:
+   - Only `opted_in_user` can receive AI-generated advice
+   - AI generated responses require disclaimers for regulatory compliance
+
+```python {title="pydantic_ai_examples/ai-access-control-config.py"}
+#! examples/pydantic_ai_examples/ai-access-control-config.py
+```
+
+## Demo
+
+Watch how PydanticAI ensures secure financial advice:
+
+![AI Access Control Live Demo](https://github.com/user-attachments/assets/e97c0a9e-d0a0-45f8-8da9-710e539d42f3)
+
+
+## Further Reading
+
+With PydanticAI's tools and structured response system with Permit.io's permission checks, AI developers can implement fine-grained AI Access Control with minimal effort.
+
+In the following blog post, we cover more in-depth topics in AI Access Control using PydanticAI and Permit.io: https://www.permit.io/blog/ai-agents-access-control-with-pydantic-ai

examples/pydantic_ai_examples/ai-access-control-config.py

@@ -0,0 +1,270 @@
+"""Permit.io configuration for secure AI Access Control Agent.
+
+Sets up the complete ABAC (Attribute-Based Access Control) model including:
+- Resources and their attributes
+- Roles and their base permissions
+- Condition sets for fine-grained access control
+- User sets with their attributes
+- Resource sets based on classification levels
+"""
+
+import asyncio
+import os
+from typing_extensions import TypedDict
+
+from permit import (
+    ConditionSetCreate,
+    Permit,
+    ResourceAttributeCreate,
+    ResourceCreate,
+    RoleCreate,
+)
+
+# API keys
+PERMIT_KEY = os.environ['PERMIT_KEY']
+
+# Initialize Permit.io SDK
+permit = Permit(
+    token=PERMIT_KEY,
+)
+
+
+# Define resources for Financial Advisor security
+class ActionConfig(TypedDict):
+    """Type definition for action configuration."""
+
+    pass  # Empty dict for actions
+
+
+class AttributeConfig(TypedDict):
+    """Type definition for attribute configuration."""
+
+    type: str
+    description: str
+
+
+class ResourceConfig(TypedDict):
+    """Type definition for resource configuration."""
+
+    key: str
+    name: str
+    description: str
+    actions: dict[str, ActionConfig]
+    attributes: dict[str, AttributeConfig]
+
+
+resources: list[ResourceConfig] = [
+    {
+        'key': 'financial_advice',
+        'name': 'Financial Advice',
+        'description': 'AI-generated financial advice',
+        'actions': {
+            'receive': {},
+        },
+        'attributes': {
+            'is_ai_generated': {
+                'type': 'bool',
+                'description': 'Whether the advice is AI-generated',
+            },
+        },
+    },
+    {
+        'key': 'financial_response',
+        'name': 'Financial Response',
+        'description': 'AI-generated response content',
+        'actions': {
+            'requires_disclaimer': {},
+        },
+        'attributes': {
+            'contains_advice': {
+                'type': 'bool',
+                'description': 'Whether the response contains financial advice',
+            },
+        },
+    },
+]
+
+# Define user attributes
+user_attributes = [
+    {
+        'key': 'ai_advice_opted_in',
+        'type': 'bool',
+        'description': 'Whether user has opted in to receive AI-generated advice',
+    },
+]
+
+# Define user sets with their attributes
+user_sets = [
+    {
+        'key': 'opted_in_users',
+        'name': 'AI Advice Opted-in Users',
+        'description': 'Users who have consented to AI-generated advice',
+        'type': 'userset',
+        'conditions': {'allOf': [{'user.ai_advice_opted_in': {'equals': True}}]},
+    },
+]
+
+# Define resource sets based on classification
+resource_sets = [
+    {
+        'key': 'finance_advice',
+        'type': 'resourceset',
+        'resource_id': 'financial_advice',
+        'name': 'Financial Advice',
+        'description': 'Financial advice with ai content',
+        'conditions': {'allOf': [{'resource.is_ai_generated': {'equals': True}}]},
+    },
+]
+
+# Define condition set rules to link user sets with resource sets
+condition_set_rules = [
+    {
+        'user_set': 'opted_in_users',
+        'permission': 'financial_advice:receive',
+        'resource_set': 'finance_advice',
+    },
+]
+
+# Define roles with ABAC rules
+roles = [
+    {'name': 'restricted_user'},
+    {
+        'name': 'premium_user',
+        'permissions': [
+            {
+                'resource': 'financial_advice',
+                'actions': ['receive'],
+                'attributes': {'is_ai_generated': ['true', 'false']},
+                'condition_sets': ['opt_in_check'],
+            },
+        ],
+    },
+]
+
+# Define example users with their attributes
+example_users = [
+    {
+        'key': 'user@example.com',
+        'email': 'user@example.com',
+        'first_name': 'Example',
+        'last_name': 'User',
+        'attributes': {
+            'ai_advice_opted_in': True,
+        },
+        'role': 'premium_user',
+    },
+    {
+        'key': 'restricted@example.com',
+        'email': 'restricted@example.com',
+        'first_name': 'Restricted',
+        'last_name': 'User',
+        'attributes': {
+            'ai_advice_opted_in': False,
+        },
+        'role': 'restricted_user',
+    },
+]
+
+
+async def create_resources(permit: Permit) -> None:
+    """Create resources in Permit.io."""
+    print('\nCreating resources...')
+    for resource in resources:
+        try:
+            print(f'\nAttempting to create resource: {resource["name"]}')
+            print(f'Resource config: {resource}')
+            await permit.api.resources.create(ResourceCreate(**resource))
+            print(f'✓ Successfully created resource: {resource["name"]}')
+        except Exception as e:
+            print(f'✗ Failed to create resource {resource["name"]}')
+            print(f'Error details: {str(e)}')
+            raise
+
+
+async def create_user_attributes(permit: Permit) -> None:
+    """Create user attributes in Permit.io."""
+    print('\nCreating user attributes...')
+    for attr in user_attributes:
+        try:
+            print(f'\nAttempting to create user attribute: {attr["key"]}')
+            print(f'Attribute config: {attr}')
+            await permit.api.resource_attributes.create(
+                '__user', ResourceAttributeCreate(**attr)
+            )
+            print(f'✓ Successfully created user attribute: {attr["key"]}')
+        except Exception as e:
+            print(f'✗ Failed to create user attribute {attr["key"]}')
+            print(f'Error details: {str(e)}')
+            raise
+
+
+async def create_roles(permit: Permit) -> None:
+    """Create roles in Permit.io."""
+    print('\nCreating roles...')
+    for role in roles:
+        role_name = str(role.get('name', ''))
+        try:
+            print(f'\nAttempting to create role: {role_name}')
+            permissions: list[str] = []
+            if isinstance(role, dict) and 'permissions' in role:
+                for permission in role['permissions']:
+                    if isinstance(permission, dict) and 'actions' in permission:
+                        actions = permission.get('actions', [])
+                        if isinstance(actions, list):
+                            for action in actions:
+                                if isinstance(permission.get('resource'), str):
+                                    permissions.append(
+                                        f'{permission["resource"]}:{action}'
+                                    )
+
+            role_data = RoleCreate(
+                name=role_name,
+                key=role_name.lower().replace(' ', '_'),
+                permissions=permissions,
+                description=f'Role for {role_name} with ABAC rules',
+            )
+            await permit.api.roles.create(role_data)
+            print(f'✓ Successfully created role: {role_name}')
+        except Exception as e:
+            print(f'✗ Failed to create role {role_name}')
+            print(f'Error details: {str(e)}')
+            raise
+
+
+async def create_condition_sets(permit: Permit) -> None:
+    """Create user and resource sets in Permit.io."""
+    print('\nCreating user and resource sets...')
+    for user_set in user_sets:
+        try:
+            await permit.api.condition_sets.create(ConditionSetCreate(**user_set))
+        except Exception as e:
+            print(f'Error creating user set: {str(e)}')
+            raise
+
+    for resource_set in resource_sets:
+        try:
+            await permit.api.condition_sets.create(ConditionSetCreate(**resource_set))
+        except Exception as e:
+            print(f'Error creating resource set: {str(e)}')
+            raise
+
+
+async def create_permit_config():
+    """Create all required configurations in Permit.io."""
+    try:
+        print('\n=== Starting Permit.io Configuration ===\n')
+
+        await create_resources(permit)
+        await create_user_attributes(permit)
+        await create_roles(permit)
+        await create_condition_sets(permit)
+
+        print('\n=== Configuration completed successfully ===\n')
+    except Exception as e:
+        print('\n✗ Configuration failed')
+        print(f'Final error: {str(e)}')
+        raise
+
+
+if __name__ == '__main__':
+    asyncio.run(create_permit_config())