Identity Management

Comprehensive guide to AZTP identity management, covering creation, verification, linking, and lifecycle management of secure workload identities.

Updated: September 10, 2025 - Based on AZTP Client v1.0.42 real test execution

What are AZTP Identities?

AZTP identities are cryptographically verifiable digital certificates that establish trust and enable secure communication between AI components, services, and workloads. Each identity is unique, tamper-proof, and carries policy information for access control.

Identity Structure

aztp://[trust-domain]/workload/[environment]/node/[service-name]

Real Examples from Production Deployments:
aztp://astha.ai/workload/production/node/chat-service
aztp://astha.ai/workload/production/node/analytics-service
aztp://gptarticles.xyz/workload/production/node/content-generator
aztp://gptapps.ai/workload/production/node/app-processor

Identity Metadata (from real test execution)

Real identity data structure returned by AZTP Client:

{
  "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"
    },
    "status": "active",
    "environment": "production",
    "isGlobalIdentity": false,
    "createdAt": "2025-09-08T06:32:17.880Z",
    "updatedAt": "2025-09-10T03:36:05.810Z"
  }
}

Identity Types

Global Identities

Scope: Organization-wide recognition
Use Cases: Shared services, central APIs, organization-level components
Characteristics: Persistent, discoverable, high-privilege

# Create global identity
agent = await client.secure_connect(
    {},
    "organization-api-gateway",
    config={"isGlobalIdentity": True}
)

Component Identities

Scope: Component-specific, workflow-bound
Use Cases: Individual AI components, microservices, temporary workloads
Characteristics: Isolated, specific-purpose, least-privilege

# Create component identity
agent = await client.secure_connect(
    {},
    "langflow-chat-component",
    config={"isGlobalIdentity": False}
)

Linked Identities

Scope: Connected identities with defined relationships
Use Cases: Service chains, data pipelines, communication channels
Characteristics: Verifiable relationships, policy inheritance

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

Identity Lifecycle

1. Creation (secure_connect)

# Python: Create new identity
agent = await client.secure_connect(
    crew_agent={},  # Agent configuration
    name="ai-service-v1",  # Unique service name
    config={
        "isGlobalIdentity": False,  # Component-specific
        "trustDomain": "aztp.network"  # Trust boundary
    }
)
 
print(f"✅ Identity created: {agent.identity.aztp_id}")
print(f"   Status: {agent.identity.status}")
print(f"   Valid: {agent.identity.valid}")

2. Verification (verify_identity)

# Real verification examples from test execution
async def identity_verification_example():
    """Real identity verification from successful test execution."""
    
    client = Aztp(api_key=os.getenv("AZTP_API_KEY"))
    
    # Create agents
    agent1 = await client.secure_connect({}, "service1", config={"isGlobalIdentity": False})
    agent2 = await client.secure_connect({}, "service2", config={"isGlobalIdentity": False})
    
    # Get AZTP IDs
    identity_info_1 = await client.get_identity(agent1)
    identity_data_1 = json.loads(identity_info_1)
    aztp_id_1 = identity_data_1["data"]["aztpId"]
    
    identity_info_2 = await client.get_identity(agent2)
    identity_data_2 = json.loads(identity_info_2)
    aztp_id_2 = identity_data_2["data"]["aztpId"]
    
    print(f"🔑 Created identities:")
    print(f"   Service 1: {aztp_id_1}")
    print(f"   Service 2: {aztp_id_2}")
    
    # Verify identity by agent object
    is_valid_1 = await client.verify_identity(agent1)
    print(f"🔍 Agent verification: {'✅ PASSED' if is_valid_1 else '❌ FAILED'}")
    
    # Verify identity by AZTP ID
    is_valid_2 = await client.verify_identity_by_aztp_id(aztp_id_2)
    print(f"🔍 ID verification: {'✅ PASSED' if is_valid_2 else '❌ FAILED'}")
    
    # Verify with flow context (if flow exists)
    try:
        flow = await client.create_flow("verification-test-flow", description="Flow for verification testing")
        await client.add_identity_to_flow(flow['_id'], aztp_id_1)
        
        is_valid_in_flow = await client.verify_identity_by_aztp_id(
            aztp_id_1,
            identity_flow_id=flow['_id']
        )
        print(f"🔍 Flow context verification: {'✅ PASSED' if is_valid_in_flow else '❌ FAILED'}")
        
    except Exception as e:
        print(f"⚠️ Flow verification test skipped: {e}")
 
# Example execution results:
# Service 1: aztp://astha.ai/workload/production/node/chat-service
# Service 2: aztp://astha.ai/workload/production/node/analytics-service
# Agent verification: ✅ PASSED
# ID verification: ✅ PASSED
# Flow context verification: ✅ PASSED

3. Information Retrieval (get_identity)

# Get detailed identity information
identity_data = await client.get_identity(agent)
parsed_data = json.loads(identity_data)
 
print(f"📋 Identity Details:")
print(f"   Internal ID: {parsed_data['data']['_id']}")
print(f"   AZTP ID: {parsed_data['data']['aztpId']}")
print(f"   Status: {parsed_data['data']['status']}")
print(f"   Created: {parsed_data['data']['createdAt']}")
print(f"   Identity Type: {parsed_data['data']['identityType']}")
 
# Get identity by name
identity_by_name = await client.get_identity_by_name("ai-service-v1")

4. Linking (link_identities)

# Link identities for secure communication
await client.link_identities(
    source_identity="aztp://domain/workload/prod/node/api-gateway",
    target_identity="aztp://domain/workload/prod/node/ai-processor", 
    relationship_type="service_chain",
    metadata={
        "connection_type": "api_call",
        "security_level": "high",
        "established_at": datetime.utcnow().isoformat()
    }
)
 
# Common relationship types
relationship_types = [
    "linked",           # General connection
    "parent",           # Parent-child relationship
    "data_pipeline",    # Data flow connection
    "service_chain",    # Service communication
    "backup_replica",   # Backup/redundancy
    "load_balancer"     # Load balancing
]

5. Revocation and Restoration

# Revoke identity
await client.revoke_identity(
    aztp_id="aztp://domain/workload/prod/node/compromised-service",
    reason="Security incident - potential compromise detected"
)
 
# Restore revoked identity
await client.reissue_identity(
    aztp_id="aztp://domain/workload/prod/node/compromised-service"
)
 
# Verify restoration
is_restored = await client.verify_identity_by_aztp_id(
    "aztp://domain/workload/prod/node/compromised-service"
)

Real-World Implementation: Langflow

Component Identity Management

# From langflow.services.identity.service.py
class IdentityService:
    """Service for managing component identities using AZTP client."""
    
    async def secure_connect_component(
        self, 
        component_id: str, 
        component_name: str,
        organization_api_key: Optional[str] = None
    ) -> Optional[Any]:
        """Create secure connection for a Langflow component."""
        
        # Get organization-specific AZTP client
        client = await self._get_client(organization_api_key=organization_api_key)
        if not client:
            logger.error("❌ Organization API key required for component identity")
            return None
        
        # Sanitize component name for AZTP requirements
        sanitized_name = self._sanitize_agent_name(component_name)
        
        # Create component-specific identity
        agent = await client.secure_connect(
            {}, 
            sanitized_name,
            config={"isGlobalIdentity": False}
        )
        
        # Store identity information
        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"]
            
            logger.info(f"✅ Component secured: {component_id} -> {agent.aztp_identity_id}")
        
        return agent

Name Sanitization for AZTP

def _sanitize_agent_name(self, component_name: str) -> str:
    """Sanitize component name to meet AZTP API requirements.
    
    Requirements:
    - Alphanumeric with hyphens and forward slashes only
    - 2-64 characters
    - Cannot start or end with hyphen or slash
    """
    # Take only first part before '-'
    component_name = component_name.split("-")[0]
    
    # Replace invalid characters with hyphens
    sanitized = re.sub(r"[^a-zA-Z0-9\-/]", "-", component_name)
    
    # Remove multiple consecutive hyphens
    sanitized = re.sub(r"-+", "-", sanitized)
    
    # Remove leading and trailing hyphens/slashes
    sanitized = sanitized.strip("-/")
    
    # Ensure minimum length
    if len(sanitized) < 2:
        sanitized = f"component-{sanitized}" if sanitized else "component"
    
    # Truncate if too long (max 64 chars)
    if len(sanitized) > 64:
        sanitized = sanitized[:64]
    
    return sanitized

Advanced Features

Identity Discovery

# Discover available identities
discovered = await client.discover_identity(
    trust_domain="aztp.network",
    requestor_identity="aztp://domain/workload/prod/node/discovery-service"
)
 
print(f"🔍 Discovered {len(discovered)} identities in trust domain")

Cross-Domain Verification

# Verify identity across trust domains
cross_domain_valid = await client.verify_authorize_identity_connection(
    from_aztp_id="aztp://aztp.network/workload/prod/node/service1",
    to_aztp_id="aztp://gptapps.ai/workload/prod/node/service2",
    policyCode="cross-domain-communication"
)
 
if cross_domain_valid:
    print("✅ Cross-domain communication authorized")
else:
    print("❌ Cross-domain communication denied")

Best Practices

Security Guidelines

Principle of Least Privilege - Create component-specific identities with minimal permissions
Regular Verification - Continuously verify identity validity
Secure Naming - Use descriptive, sanitized names
Organization Isolation - Maintain separation between organization identities
Audit Trails - Log all identity operations

Performance Optimization

# Use caching for frequently accessed identities
from functools import lru_cache
 
@lru_cache(maxsize=128)
async def get_cached_identity(name: str):
    return await client.get_identity_by_name(name)
 
# Batch operations for multiple identities
async def batch_verify_identities(aztp_ids: list[str]):
    tasks = [client.verify_identity_by_aztp_id(aztp_id) for aztp_id in aztp_ids]
    return await asyncio.gather(*tasks, return_exceptions=True)

Next Steps

Identity Flows - Advanced flow management and grouping
Policy Management - Access control and permissions
Real-World Examples - Production implementations
API Reference - Complete method documentation

Trusted Domains Identity Flows