Proposal: APort Agent Passport Middleware - Pre-execution Authorization for Microsoft Agent Framework

[King Uchi (uchibeke)]

🛡️ APort Middleware Integration for Microsoft Agent Framework

Overview

APort provides a universal, open, enterprise-grade guardrail, identity verification, policy enforcement, and audit trails for AI agents. We propose integrating APort as middleware for Microsoft Agent Framework to enable pre-execution authorization - ensuring agents are verified and compliant before they execute any actions.

🎯 Value Proposition for Microsoft Agent Framework

Current Challenge

• Agents execute actions with no easy way do handle pre-authorization checks
• No standardized way to verify agent identity before execution
• Limited audit trails for compliance (SOC 2, IIROC, OSFI)
• Risk of unauthorized agent actions in enterprise environments

APort Solution

• Pre-execution verification: Verify agent passports before agent runs
• Policy enforcement: Enforce spending limits, data access, regional restrictions
• Real-time authorization: ms authorization decisions
• Audit trails: Cryptographically signed attestations for compliance
• Kill switch: Global agent suspension capability

🔧 Technical Integration

Proposed Implementation

Based on the Microsoft Agent Framework middleware documentation, we'll use the function-based middleware approach for simplicity and elegance:

from typing import Callable, Awaitable, Optional, Dict, Any
from agent_framework import AgentRunContext
from aporthq_sdk_python import APortClient, APortClientOptions, AportError, PolicyVerificationResponse

async def aport_agent_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    """APort middleware for agent authorization and policy enforcement."""
    
    # Initialize APort client (in production, this would be injected)
    client = APortClient(APortClientOptions(
        base_url="https://api.aport.io", # Optional
        api_key="your-api-key",  #  From environment or config. Optional. No API key or account required.
        timeout_ms=800  # Optional
    ))
    
    # Extract agent ID from context metadata
    agent_id = context.metadata.get('agent_id') or context.metadata.get('agent_passport_id')
    
    if not agent_id:
        # Fail closed - terminate execution if no agent ID provided
        context.terminate = True
        context.result = {
            "error": "missing_agent_id",
            "message": "Agent ID is required for authorization"
        }
        return
    
    try:
        # Extract policy ID from context metadata
        policy_id = context.metadata.get('policy_id')
        
        if policy_id:
            # Policy verification includes passport verification automatically
            decision: PolicyVerificationResponse = await client.verify_policy(
                agent_id=agent_id,
                policy_id=policy_id,
                context=_extract_context_data(context)
            )

            # https://github.com/aporthq/aport-spec/blob/main/oap/decision-schema.json
            if not decision.allow:
                # Policy violation - terminate execution
                context.terminate = True
                context.result = {
                    "error": "policy_violation",
                    "decision_id": decision.decision_id,
                    "reasons": decision.reasons,
                    "assurance_level": decision.assurance_level
                }
                return
                
            # Store policy decision in context for audit trail
            context.metadata['aport_decision'] = {
                "decision_id": decision.decision_id,
                "allow": decision.allow,
                "assurance_level": decision.assurance_level,
                "reasons": decision.reasons
            }
        else:
            # Only verify passport if no policy specified
            passport_view = await client.get_passport_view(agent_id)
            context.metadata['agent_passport'] = passport_view
        
        # Continue to next middleware or agent execution
        await next(context)
        
        # Generate audit trail after successful execution
        if context.result and not context.terminate:
            await _generate_audit_trail(client, agent_id, context)
            
    except AportError as e:
        # APort API error - terminate execution
        context.terminate = True
        context.result = {
            "error": "agent_verification_failed",
            "status": e.status,
            "message": str(e),
            "reasons": e.reasons,
            "decision_id": e.decision_id
        }
    except Exception as e:
        # Unexpected error - terminate execution
        context.terminate = True
        context.result = {
            "error": "internal_error",
            "message": f"Authorization failed: {str(e)}"
        }
    finally:
        # Clean up client resources
        await client.close()


def _extract_context_data(context: AgentRunContext) -> Dict[str, Any]:
    """Extract relevant context data for policy evaluation."""
    return {
        "action": context.metadata.get('action', 'unknown'),
        "resource": context.metadata.get('resource'),
        "amount": context.metadata.get('amount'),
        "region": context.metadata.get('region'),
        "timestamp": context.metadata.get('timestamp'),
        # Add any other context data needed for policy evaluation
    }


async def _generate_audit_trail(
    client: APortClient, 
    agent_id: str, 
    context: AgentRunContext
) -> None:
    """Generate cryptographically signed audit trail."""
    audit_data = {
        "agent_id": agent_id,
        "action": context.metadata.get('action', 'unknown'),
        "result": context.result,
        "timestamp": context.metadata.get('timestamp'),
        "policy_id": context.metadata.get('policy_id'),
        "decision_id": context.metadata.get('aport_decision', {}).get('decision_id')
    }
    
    # Note: This would be implemented when audit trail API is available
    # await client.create_audit_trail(audit_data)
    print(f"Audit trail: {audit_data}")  # Placeholder

Usage Example

from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

async def main():
    credential = AzureCliCredential()
    
    # Create agent with APort middleware
    async with AzureAIAgentClient(async_credential=credential).create_agent(
        name="RefundAgent",
        instructions="You are a helpful refund assistant.",
        tools=[process_refund_tool],
        middleware=[aport_agent_middleware],  # Agent-level middleware
    ) as agent:
        
        # Run with agent ID and policy context
        result = await agent.run(
            "Process a $50 refund for order 12345",
            middleware=[],  # No additional run-level middleware needed
            metadata={
                'agent_id': 'ap_a2d10232c6534523812423eec8a1425c45678',
                'policy_id': 'finance.payment.refund.v1',
                'action': 'refund',
                'amount': 5000,  # Amount in cents
                'region': 'US'
            }
        )
        
        print(f"Refund result: {result}")


# Tool function that benefits from pre-authorization
async def process_refund_tool(order_id: str, amount: int) -> str:
    """Process a refund - only called if APort authorization passes."""
    return f"Refund of ${amount/100:.2f} processed for order {order_id}"

Alternative: Class-Based Implementation

For more complex scenarios requiring stateful operations:

from agent_framework import AgentMiddleware, AgentRunContext
from aporthq_sdk_python import APortClient, APortClientOptions, AportError

class AportAgentMiddleware(AgentMiddleware):
    """Class-based APort middleware for agent authorization."""
    
    def __init__(self, api_key: str, base_url: str = "https://api.aport.io"):
        self.client_options = APortClientOptions(
            base_url=base_url,
            api_key=api_key,
            timeout_ms=800
        )
    
    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        """Process agent run with APort authorization."""
        
        async with APortClient(self.client_options) as client:
            agent_id = context.metadata.get('agent_id')
            
            if not agent_id:
                context.terminate = True
                context.result = {"error": "missing_agent_id"}
                return
            
            try:
                policy_id = context.metadata.get('policy_id')
                
                if policy_id:
                    decision = await client.verify_policy(
                        agent_id=agent_id,
                        policy_id=policy_id,
                        context=self._extract_context(context)
                    )
                    
                    if not decision.allow:
                        context.terminate = True
                        context.result = {
                            "error": "policy_violation",
                            "decision_id": decision.decision_id,
                            "reasons": decision.reasons
                        }
                        return
                
                await next(context)
                
            except AportError as e:
                context.terminate = True
                context.result = {
                    "error": "agent_verification_failed",
                    "status": e.status,
                    "message": str(e),
                    "reasons": e.reasons
                }
    
    def _extract_context(self, context: AgentRunContext) -> Dict[str, Any]:
        """Extract context data for policy evaluation."""
        return {
            "action": context.metadata.get('action', 'unknown'),
            "amount": context.metadata.get('amount'),
            "region": context.metadata.get('region')
        }

🏢 Enterprise Use Cases

Financial Services

• Trading Agents: Prevent unauthorized trading actions
• Payment Agents: Enforce spending limits and regional restrictions
• Compliance: Ensure adherence to IIROC, OSFI regulations

Healthcare

• Patient Data Agents: Verify access to PHI (Protected Health Information)
• HIPAA Compliance: Enforce data access policies
• Audit Requirements: Complete audit trails for regulatory compliance

E-commerce

• Refund Agents: Prevent fraudulent refund requests
• Inventory Agents: Enforce inventory management policies
• Fraud Prevention: Real-time authorization for high-value transactions

📊 Benefits for Microsoft Agent Framework Ecosystem

1. Enterprise Adoption: Enables enterprise customers to deploy agents with confidence
2. Compliance: Built-in SOC 2, IIROC, OSFI compliance features
3. Security: Prevents unauthorized agent actions with pre-execution checks
4. Auditability: Complete audit trails for regulatory requirements
5. Policy Management: Centralized policy enforcement across all agents
6. Scalability: Handles thousands of concurrent agent verifications
7. Multi-region: Passports and Policy verification can be handled in a region, for compliance
8. Private instance: Ability to deploy Aport on-prem in private cloud or servers

🚀 Implementation Approach

Phase 1: Core Middleware (First PR)

• Agent Run Middleware for passport verification
• Basic policy enforcement with common policies
• Audit trail generation and storage
• Error handling and logging
• Agent Passport Creation on Aport
• Kill switch functionality for emergency agent suspension

Phase 2: Advanced and Enterprise Features (2nd PR)

• Function Calling Middleware for tool-level authorization
• Performance optimizations
• Advanced compliance reporting and dashboards
• Multi-tenant support and organization management

🔧 Technical Specifications

Performance Requirements

• Authorization Latency: Targeting <100ms (95th percentile)
• Availability: 99.9% uptime SLA
• Caching: Caching for passport data

Security Features

• Cryptographic Signatures: All audit trails are cryptographically signed
• Rate Limiting: Built-in rate limiting to prevent abuse

🤝 Community Benefits

• Developers: Easy-to-use middleware for agent security with minimal code changes
• Enterprises: Compliance-ready agent deployments out of the box
• Microsoft: Enhanced enterprise adoption of Agent Framework
• Ecosystem: Standardized agent authorization patterns and best practices

📋 Questions for Discussion

1. Architecture: Is Agent Run Middleware the optimal integration point for pre-execution authorization?
2. Implementation: Should this be a separate package (@aporthq/microsoft-agent-middleware implement at https://github.com/aporthq/aport-sdks-and-middlewares) or integrated into the core framework?
3. Policies: How should policy definitions be managed, versioned, and distributed across the ecosystem? Currently, they are at https://github.com/aporthq/aport-policies
4. Performance: What are the acceptable latency requirements for authorization in different use cases?
5. Error Handling: How should authorization failures be handled in different agent execution contexts? Currently, we have Open Agent Passport (OAP) Spec, https://github.com/aporthq/aport-spec, that defines decision, errors etc

🔗 Resources

• Live Demo: demo.aport.io - Interactive APort demonstration
• Documentation: aport.io/docs - Complete technical documentation
• GitHub: github.com/aporthq - Open source implementation

💡 Next Steps

1. Community Feedback: Gather input on integration approach and requirements
2. Proof of Concept: Build initial middleware implementation with Microsoft Agent Framework
3. Pull Request: Submit PR with working middleware and comprehensive tests
4. Documentation: Create integration guides, examples, and best practices
5. Discord Launch: Share implementation on Microsoft Agent Framework Discord community

---

We're excited to contribute to the Microsoft Agent Framework ecosystem and help make AI agents more secure and compliant for enterprise use cases!

What are your thoughts on this integration approach? We'd love to hear from the community and Microsoft team about the technical implementation and enterprise requirements, before we create an issue and PR