API Reference

from aztp_client import Aztp
 
# Basic initialization
client = Aztp(api_key="your-api-key")
 
# With custom base URL
client = Aztp(
    api_key="your-api-key",
    base_url="https://your-aztp-server.com"
)

import { AztpClient } from 'aztp-client';
 
// Basic initialization
const client = new AztpClient('your-api-key');
 
// With custom configuration
const client = new AztpClient('your-api-key', {
  baseUrl: 'https://your-aztp-server.com'
});

# Python: Create component identity
agent = await client.secure_connect(
    {}, 
    "langflow-chat-component",
    config={"isGlobalIdentity": False}
)
print(f"Identity created: {agent.identity.aztp_id}")

// TypeScript: Create service identity
const agent = await client.secureConnect(
  {}, 
  'api-gateway-service',
  { isGlobalIdentity: true }
);
console.log(`Identity created: ${agent.identity.aztpId}`);

# Python: Verify component identity
is_valid = await client.verify_identity(agent)
if is_valid:
    print("Identity verification successful")
else:
    print("Identity verification failed")
 
# Verify within flow context
is_valid_in_flow = await client.verify_identity(
    agent, 
    identity_flow_id="flow-uuid"
)

# Python: Verify by AZTP ID
aztp_id = "aztp://domain/workload/prod/node/service1"
is_valid = await client.verify_identity_by_aztp_id(aztp_id)
 
# With flow context
is_valid_in_flow = await client.verify_identity_by_aztp_id(
    aztp_id,
    identity_flow_id="flow-uuid"
)

# Python: Get identity details
identity_data = await client.get_identity(agent)
parsed_data = json.loads(identity_data)
 
print(f"Identity ID: {parsed_data['data']['_id']}")
print(f"AZTP ID: {parsed_data['data']['aztpId']}")
print(f"Status: {parsed_data['data']['status']}")
print(f"Environment: {parsed_data['data']['environment']}")
print(f"Trust Domain: {parsed_data['data']['workloadInfo']['trustDomain']}")

{
  "success": true,
  "message": "WI-FND-S-001 - Workload identity found successfully",
  "data": {
    "_id": "507f1f77bcf86cd799439011",
    "userId": "507f1f77bcf86cd799439012",
    "aztpId": "aztp://astha.ai/workload/production/node/chat-service",
    "name": "chat-service",
    "parentIdentity": null,
    "absolutePath": "",
    "selectors": [
      "workload:id:chat-service",
      "agent:id:aztp",
      "method:node",
      "metadata:hostname:ai-production-server",
      "metadata:environment:production",
      "metadata:trustDomain:astha.ai",
      "metadata:parentIdentity:null",
      "metadata:isFree:true"
    ],
    "workloadInfo": {
      "workloadId": "chat-service",
      "agentId": "aztp",
      "trustDomain": "astha.ai",
      "method": "node",
      "timestamp": "2025-09-10T03:36:05.794Z",
      "hostname": "ai-production-server",
      "environment": "production",
      "additionalMetadata": {
        "hostname": "ai-production-server",
        "environment": "production",
        "trustDomain": "astha.ai",
        "parentIdentity": null,
        "isFree": true,
        "linkTo": []
      }
    },
    "issuedAt": 1757313137864,
    "rotateAt": 1757561765793,
    "metadata": {
      "hostname": "ai-production-server",
      "environment": "production",
      "trustDomain": "astha.ai",
      "parentIdentity": null,
      "isFree": true,
      "linkTo": []
    },
    "status": "active",
    "environment": "production",
    "expiresAt": 1757561765793,
    "rotationCount": 6,
    "available": true,
    "isPaid": false,
    "isFree": true,
    "isGlobalIdentity": false,
    "isGloballyDiscoverable": false,
    "createdAt": "2025-09-08T06:32:17.880Z",
    "updatedAt": "2025-09-10T03:36:05.810Z"
  }
}

# Python: Link identities
await client.link_identities(
    "aztp://domain/workload/prod/node/service1",
    "aztp://domain/workload/prod/node/service2",
    "linked",
    metadata={"connection_type": "data_flow"}
)

# Python: Get policy
policy_data = await client.get_policy(
    "aztp://astha.ai/workload/production/node/chat-service"
)
 
print(f"Retrieved {len(policy_data)} policies")
for policy in policy_data:
    print(f"Policy Code: {policy['code']}")
    print(f"Actions: {policy['policyStatement']['Statement']['Action']}")

# Real policy structure returned by get_policy()
[
  {
    'policyStatement': {
      'Version': '2025-09-10',
      'Statement': {
        'Sid': '74298e6a-d44d-421f-843e-ec52f4ef7bea',
        'Effect': 'Allow',
        'Action': ['read_user_profile', 'list_users'],
        'Condition': {
          'StringEquals': {
            'department': 'Engineering',
            'trust_domain': 'astha.ai'
          }
        }
      }
    },
    'code': 'policy:4f4b07f75458',
    'identity': {
      'name': 'service1',
      'aztpId': 'aztp://astha.ai/workload/production/node/service1'
    }
  },
  {
    'policyStatement': {
      'Version': '2025-09-10',
      'Statement': {
        'Sid': '74298e6a-d44d-421f-843e-ec52f4ef7bea',
        'Effect': 'Allow',
        'Action': ['read_user_profile', 'list_users', 'test'],
        'Condition': {
          'StringEquals': {
            'department': 'Engineering',
            'trust_domain': 'astha.ai'
          }
        }
      }
    },
    'code': 'policy:292a1ea03cbc',
    'identity': {
      'name': 'service1',
      'aztpId': 'aztp://astha.ai/workload/production/node/service1'
    }
  }
]

# Python: Filter policy by code
policies = await client.get_policy(aztp_id)
specific_policy = client.get_policy_value(
    policies=policies,
    filter_key="code",
    filter_value="policy:4f4b07f75458"
)
 
# Real filtered policy result:
# {
#   'Version': '2025-09-10',
#   'Statement': {
#     'Effect': 'Allow',
#     'Action': ['read_user_profile', 'list_users'],
#     'Condition': {'StringEquals': {'department': 'Engineering'}}
#   }
# }

# Python: Check if action is allowed
policy_statement = client.get_policy_value(policies, "code", "policy:292a1ea03cbc")
if policy_statement:
    is_allowed = client.is_action_allowed(policy_statement, "test")
    print(f"Action 'test' allowed: {is_allowed}")  # True for policy:292a1ea03cbc
 
# Test multiple actions
test_actions = ["read", "write", "execute", "delete", "test", "admin"]
for action in test_actions:
    is_allowed = client.is_action_allowed(policy_statement, action)
    status = "✅ ALLOWED" if is_allowed else "❌ DENIED"
    print(f"Action '{action}': {status}")
 
# Real test results from execution:
# Action 'read': ❌ DENIED
# Action 'write': ❌ DENIED
# Action 'execute': ❌ DENIED
# Action 'delete': ❌ DENIED
# Action 'test': ✅ ALLOWED  # Only for policy:292a1ea03cbc
# Action 'admin': ❌ DENIED

# Python: Check permissions for specific actions
permissions = await client.check_identity_policy_permissions(
    "aztp://domain/workload/prod/node/service1",
    options={"actions": ["read", "write", "api_call"]}
)
 
print(f"Read permission: {permissions.get('read', False)}")
print(f"Write permission: {permissions.get('write', False)}")
print(f"API call permission: {permissions.get('api_call', False)}")

# Python: OIAP evaluation (working example)
result = await client.oiap_evaluate(
    aztp_id="aztp://astha.ai/workload/production/node/chat-service",
    requested_resource="list_users",
    user_id="user-demo-12345"
)
 
print(f"OIAP Result: {result['result']}")
print(f"Reason: {result.get('reason', 'No reason provided')}")
 
# Test multiple resources
oiap_test_cases = [
    {"resource": "list_users", "user_id": "user-demo-12345"},
    {"resource": "database/users", "user_id": "user-demo-67890"},
    {"resource": "api/admin/settings", "user_id": "admin-demo-456"},
    {"resource": "files/documents", "user_id": "employee-demo-789"},
    {"resource": "service/public", "user_id": "guest-demo-001"}
]
 
for test_case in oiap_test_cases:
    try:
        result = await client.oiap_evaluate(
            aztp_id="aztp://astha.ai/workload/production/node/chat-service",
            requested_resource=test_case['resource'],
            user_id=test_case['user_id']
        )
        print(f"Resource '{test_case['resource']}': {result.get('result', 'UNKNOWN')}")
    except Exception as e:
        print(f"Resource '{test_case['resource']}': ERROR - {e}")

// Successful OIAP evaluation
{
  "result": "ALLOW",
  "reason": "Explicitly allowed by policy"
}
 
// Failed OIAP evaluation  
{
  "response": "Failed to evaluate OIAP policy. {{error}}",
  "status": 404,
  "message": "Failed to evaluate OIAP policy. {{error}}",
  "name": "HttpException"
}

Resource: list_users
  User ID: user-demo-12345
  Result: ALLOW
  Reason: Explicitly allowed by policy

Resource: database/users
  User ID: user-demo-67890
  Result: ERROR - Failed to evaluate OIAP policy

Resource: api/admin/settings
  User ID: admin-demo-456
  Result: ERROR - Failed to evaluate OIAP policy

# Python: Create flow (real working example)
flow = await client.create_flow(
    name="production-flow",
    description="Production deployment workflow",
    discoverable="private",
    tags=["production", "critical", "env:prod"],
    metadata={
        "environment": "production",
        "project": "aztp-testing",
        "security_level": "high",
        "astha-flow": {
            "id": "flow_123",
            "folderId": "folder_456"
        }
    }
)
print(f"Flow created: {flow['_id']}")
print(f"Flow name: {flow['name']}")
print(f"Flow tags: {flow['tags']}")

{
  "_id": "68c0f232e60425bee4e3e4d9",
  "name": "production-flow",
  "description": "Production deployment workflow",
  "tags": ["production", "critical", "env:prod"],
  "owner": "6832cc64f39bb463f2e0767e",
  "creator": "Astha.ai",
  "visibilityScope": "organization",
  "discoverable": "private",
  "metadata": {
    "environment": "production",
    "project": "aztp-testing",
    "security_level": "high",
    "astha-flow": {
      "id": "flow_123",
      "folderId": "folder_456"
    }
  },
  "createdAt": "2025-09-10T03:36:18.478Z",
  "updatedAt": "2025-09-10T03:36:18.478Z",
  "identities": []
}

# Python: Add identity to flow (working example)
flow_id = "507f1f77bcf86cd799439013"  # From create_flow response
aztp_id = "aztp://astha.ai/workload/production/node/chat-service"
 
result = await client.add_identity_to_flow(flow_id, aztp_id)
 
print(f"✅ Added identity to flow")
print(f"Flow now contains {len(result.get('identities', []))} identities")

{
  "_id": "507f1f77bcf86cd799439013",
  "name": "development-flow",
  "description": "Development environment workflow",
  "tags": ["development", "testing", "env:dev"],
  "owner": "507f1f77bcf86cd799439012",
  "creator": "Demo User",
  "visibilityScope": "organization",
  "discoverable": "public",
  "metadata": {
    "environment": "development",
    "project": "aztp-testing",
    "created_by": "notebook_demo",
    "version": "1.0"
  },
  "createdAt": "2025-09-10T03:36:18.079Z",
  "updatedAt": "2025-09-10T03:36:19.313Z",
  "identities": [
    {
      "id": "507f1f77bcf86cd799439016",
      "identity_id": "507f1f77bcf86cd799439011",
      "aztpId": "aztp://astha.ai/workload/production/node/chat-service"
    }
  ]
}

# Python: Flow-specific identity revocation
revoke_result = await client.revoke_flow_identity(
    aztp_id="aztp://astha.ai/workload/production/node/analytics-service",
    identity_flow_id="507f1f77bcf86cd799439013",
    reason="Testing flow-specific revocation"
)
 
print(f"Revocation result: {revoke_result}")
 
# Verify revocation in flow context
try:
    is_valid_in_flow = await client.verify_identity_by_aztp_id(
        aztp_id,
        identity_flow_id=flow_id
    )
    print(f"Valid in flow: {is_valid_in_flow}")  # Should be False
except Exception as e:
    if "Identity already revoked for this flow" in str(e):
        print("✅ Identity successfully revoked from flow")
 
# Check global validity (should still be True)
is_globally_valid = await client.verify_identity_by_aztp_id(aztp_id)
print(f"Globally valid: {is_globally_valid}")  # Should be True

{
  "valid": true,
  "issuedAt": 1757475385308,
  "aztpId": "aztp://astha.ai/workload/production/node/analytics-service",
  "reason": "Testing flow-specific revocation in demo"
}

# From Langflow's IdentityService implementation
class IdentityService:
    async def secure_connect_component(
        self, 
        component_id: str, 
        component_name: str,
        organization_api_key: str = None
    ):
        client = await self._get_client(organization_api_key=organization_api_key)
        
        # Create secure connection with component-specific identity
        agent = await client.secure_connect(
            {}, 
            self._sanitize_agent_name(component_name),
            config={"isGlobalIdentity": False}
        )
        
        # Store identity for policy checks
        if agent.identity.valid:
            identity_data = await client.get_identity(agent)
            parsed_data = json.loads(identity_data)
            agent.aztp_identity_id = parsed_data["data"]["aztpId"]
            agent.aztp_identity_status = parsed_data["data"]["status"]
        
        return agent

# Real implementation from Langflow
async def check_oiap_policy(self, tool_name: str, component_name: str) -> bool:
    user_id = get_current_user_id()
    organization_api_key = get_current_organization_api_key()
    organization_api_access = get_current_organization_api_access()
    
    client = await self._get_client(organization_api_key=organization_api_key)
    
    # Create agent identity for component
    agent_id = await client.secure_connect(
        {}, 
        f"{organization_api_access}/{component_name}", 
        config={"isGlobalIdentity": False}
    )
    
    # Load flow context
    with open(f"aztp_flow_id_{organization_api_key[:8]}.json", "r") as f:
        aztp_flow_id = json.load(f)
    
    # Add identity to flow
    await client.add_identity_to_flow(aztp_flow_id, agent_id.identity.aztp_id)
    
    # Evaluate OIAP policy
    result = await client.oiap_evaluate(
        aztp_id=agent_id.identity.aztp_id,
        requested_resource=tool_name,
        user_id=user_id
    )
    
    return result['result'] == "ALLOW"

try:
    agent = await client.secure_connect({}, "service1")
except ValueError as ve:
    # Validation errors (invalid parameters, trust domain issues)
    logger.error(f"Validation error: {ve}")
except ConnectionError as ce:
    # Network connectivity issues
    logger.error(f"Connection error: {ce}")
except AuthenticationError as ae:
    # API key or authentication failures
    logger.error(f"Authentication error: {ae}")
except Exception as e:
    # Unexpected errors
    logger.error(f"Unexpected error: {e}")

async def robust_identity_creation(client, name):
    """Create identity with comprehensive error handling."""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            agent = await client.secure_connect({}, name)
            is_valid = await client.verify_identity(agent)
            
            if is_valid:
                return agent
            else:
                logger.warning(f"Identity verification failed for {name}")
                
        except ConnectionError:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise
        except ValueError as e:
            logger.error(f"Invalid parameters for {name}: {e}")
            return None
    
    return None