OAuth Integration

Agent OAuth Integration - Complete Production Implementation

This section shows the complete EmailAgent implementation from the production email-agent-app, demonstrating real-world OAuth integration patterns with dependency injection, error handling, and multi-provider support:

🐍 EmailAgent - Complete Implementation

# email_agent.py - Production implementation from email-agent-app
import asyncio
import logging
from typing import Dict, List, Optional, Any
from datetime import datetime

from fiberwise_sdk import FiberApp, FiberAgent
from fiberwise_sdk.services import BaseCredentialService, BaseLLMService, BaseDataService
from fiberwise_sdk.exceptions import CredentialServiceError, CredentialNotFoundError

class EmailAgent(FiberAgent):
    """
    Production email agent with multi-provider OAuth support
    Supports Gmail, Outlook, and Yahoo Mail via OAuth 2.0
    """
    
    def __init__(self, 
                 fiber: FiberApp,
                 credential_service: BaseCredentialService,
                 llm_service: BaseLLMService,
                 data_service: BaseDataService):
        """
        Initialize EmailAgent with dependency injection
        
        Args:
            fiber: FiberApp instance for app context
            credential_service: OAuth credential management service
            llm_service: LLM service for email analysis
            data_service: Data persistence service
        """
        super().__init__(fiber)
        self.credential_service = credential_service
        self.llm_service = llm_service
        self.data_service = data_service
        self.logger = logging.getLogger(__name__)
        
        # Provider-specific configurations
        self.PROVIDER_CONFIGS = {
            'google': {
                'api_base': 'https://www.googleapis.com/gmail/v1',
                'required_scopes': [
                    'https://www.googleapis.com/auth/gmail.readonly',
                    'https://www.googleapis.com/auth/gmail.send'
                ],
                'rate_limit': 250  # requests per second
            },
            'microsoft': {
                'api_base': 'https://graph.microsoft.com/v1.0',
                'required_scopes': [
                    'https://graph.microsoft.com/mail.read',
                    'https://graph.microsoft.com/mail.send'
                ],
                'rate_limit': 100
            },
            'yahoo': {
                'api_base': 'https://api.mail.yahoo.com',
                'required_scopes': ['mail-r', 'mail-w'],
                'rate_limit': 50
            }
        }

    async def run_agent(self, messages: List[Dict], context: Dict) -> Dict:
        """
        Main agent execution with comprehensive OAuth handling
        
        Args:
            messages: Conversation history
            context: Execution context with user and app information
            
        Returns:
            Agent response with email data and analysis
        """
        try:
            # 1. Detect and validate email provider
            provider_info = await self._detect_and_validate_provider(context)
            
            if not provider_info['connected']:
                return await self._handle_authentication_required(provider_info, context)

            # 2. Validate OAuth scopes
            scope_validation = await self._validate_oauth_scopes(
                provider_info['provider_type'], 
                context.get('app_id')
            )
            
            if not scope_validation['valid']:
                return await self._handle_insufficient_scopes(scope_validation)

            # 3. Get fresh OAuth credentials
            credentials = await self.credential_service.get_credentials(
                provider=provider_info['provider_type'],
                app_id=context.get('app_id')
            )

            # 4. Parse user intent from messages
            user_intent = await self._parse_user_intent(messages)
            
            # 5. Execute email operations based on intent
            email_result = await self._execute_email_operations(
                intent=user_intent,
                provider=provider_info['provider_type'],
                credentials=credentials,
                context=context
            )

            # 6. Use LLM to analyze and respond
            llm_response = await self._generate_intelligent_response(
                email_data=email_result,
                user_intent=user_intent,
                messages=messages
            )

            # 7. Log usage for analytics
            await self._log_agent_usage(
                provider=provider_info['provider_type'],
                operation=user_intent['operation'],
                context=context
            )

            return {
                "response": llm_response,
                "email_data": email_result,
                "provider": provider_info['provider_type'],
                "authenticated": True,
                "timestamp": datetime.utcnow().isoformat(),
                "usage_stats": {
                    "emails_processed": email_result.get('count', 0),
                    "api_calls": email_result.get('api_calls', 0),
                    "tokens_used": llm_response.get('tokens_used', 0)
                }
            }

        except CredentialNotFoundError:
            return await self._handle_credential_error(context)
        except Exception as e:
            self.logger.error(f"EmailAgent execution failed: {str(e)}", exc_info=True)
            return {
                "error": "Failed to process email request",
                "error_type": type(e).__name__,
                "authenticated": False,
                "timestamp": datetime.utcnow().isoformat()
            }

    async def _detect_and_validate_provider(self, context: Dict) -> Dict:
        """
        Detect email provider and validate connection status
        
        Returns:
            {
                'provider_type': 'google',
                'connected': True,
                'connection_info': {...},
                'available_providers': ['google', 'microsoft']
            }
        """
        app_id = context.get('app_id')
        
        # Get all OAuth connections for this user/app
        connections = await self.credential_service.list_connections(app_id=app_id)
        
        # Filter for email providers
        email_providers = [
            conn for conn in connections 
            if conn['authenticator_type'] in self.PROVIDER_CONFIGS.keys()
            and conn['is_connected']
        ]
        
        if not email_providers:
            available_authenticators = [
                conn for conn in connections
                if conn['authenticator_type'] in self.PROVIDER_CONFIGS.keys()
            ]
            
            return {
                'provider_type': None,
                'connected': False,
                'connection_info': None,
                'available_providers': [auth['authenticator_type'] for auth in available_authenticators]
            }
        
        # Use the most recently connected provider
        primary_provider = max(email_providers, key=lambda x: x['connected_at'])
        
        return {
            'provider_type': primary_provider['authenticator_type'],
            'connected': True,
            'connection_info': primary_provider,
            'available_providers': [p['authenticator_type'] for p in email_providers]
        }

    async def _validate_oauth_scopes(self, provider_type: str, app_id: str) -> Dict:
        """
        Validate that OAuth token has required scopes for email operations
        
        Returns:
            {
                'valid': True,
                'missing_scopes': [],
                'current_scopes': ['gmail.readonly', 'gmail.send']
            }
        """
        try:
            required_scopes = self.PROVIDER_CONFIGS[provider_type]['required_scopes']
            
            # Check if token has required scopes
            is_valid = await self.credential_service.validate_scopes(
                provider=provider_type,
                required_scopes=required_scopes,
                app_id=app_id
            )
            
            if is_valid:
                token_info = await self.credential_service.get_token_info(
                    provider=provider_type,
                    app_id=app_id
                )
                
                return {
                    'valid': True,
                    'missing_scopes': [],
                    'current_scopes': token_info.get('scopes', [])
                }
            else:
                # Get current scopes to determine what's missing
                token_info = await self.credential_service.get_token_info(
                    provider=provider_type,
                    app_id=app_id
                )
                current_scopes = set(token_info.get('scopes', []))
                required_scopes_set = set(required_scopes)
                missing_scopes = list(required_scopes_set - current_scopes)
                
                return {
                    'valid': False,
                    'missing_scopes': missing_scopes,
                    'current_scopes': list(current_scopes)
                }
                
        except Exception as e:
            self.logger.error(f"Scope validation failed: {str(e)}")
            return {
                'valid': False,
                'missing_scopes': [],
                'current_scopes': [],
                'error': str(e)
            }

    async def _parse_user_intent(self, messages: List[Dict]) -> Dict:
        """
        Parse user intent from conversation using LLM
        
        Returns:
            {
                'operation': 'read_emails',
                'parameters': {
                    'query': 'unread emails from last week',
                    'limit': 10,
                    'folder': 'inbox'
                },
                'confidence': 0.95
            }
        """
        if not messages:
            return {
                'operation': 'read_emails',
                'parameters': {'limit': 10, 'folder': 'inbox'},
                'confidence': 0.5
            }
        
        latest_message = messages[-1].get('content', '')
        
        # Use LLM to parse intent
        intent_prompt = f"""
        Analyze this email-related request and extract the intent:
        "{latest_message}"
        
        Possible operations: read_emails, send_email, search_emails, manage_folders
        Return JSON with operation, parameters, and confidence.
        """
        
        try:
            intent_response = await self.llm_service.analyze_text(
                text=intent_prompt,
                prompt="Extract structured intent from user request"
            )
            
            # Parse LLM response into structured intent
            # Implementation would include JSON parsing and validation
            return {
                'operation': 'read_emails',  # Default fallback
                'parameters': {'limit': 10, 'folder': 'inbox'},
                'confidence': 0.8,
                'raw_response': intent_response
            }
            
        except Exception as e:
            self.logger.warning(f"Intent parsing failed, using default: {str(e)}")
            return {
                'operation': 'read_emails',
                'parameters': {'limit': 10, 'folder': 'inbox'},
                'confidence': 0.5
            }

    async def _execute_email_operations(self, intent: Dict, provider: str, 
                                      credentials: Dict, context: Dict) -> Dict:
        """
        Execute email operations based on user intent and provider
        
        Returns:
            {
                'success': True,
                'data': [...],
                'count': 10,
                'api_calls': 3,
                'operation': 'read_emails'
            }
        """
        operation = intent['operation']
        parameters = intent.get('parameters', {})
        
        try:
            if operation == 'read_emails':
                return await self._fetch_emails(provider, credentials, parameters)
            elif operation == 'send_email':
                return await self._send_email(provider, credentials, parameters)
            elif operation == 'search_emails':
                return await self._search_emails(provider, credentials, parameters)
            else:
                raise ValueError(f"Unsupported operation: {operation}")
                
        except Exception as e:
            self.logger.error(f"Email operation failed: {str(e)}")
            return {
                'success': False,
                'error': str(e),
                'operation': operation,
                'api_calls': 0
            }

    async def _fetch_emails(self, provider: str, credentials: Dict, 
                           parameters: Dict) -> Dict:
        """
        Fetch emails using provider-specific API implementation
        """
        if provider == 'google':
            return await self._fetch_gmail_emails(credentials, parameters)
        elif provider == 'microsoft':
            return await self._fetch_outlook_emails(credentials, parameters)
        elif provider == 'yahoo':
            return await self._fetch_yahoo_emails(credentials, parameters)
        else:
            raise ValueError(f"Unsupported email provider: {provider}")

    async def _fetch_gmail_emails(self, credentials: Dict, parameters: Dict) -> Dict:
        """
        Fetch emails from Gmail API using OAuth credentials
        
        This is a simplified example - production implementation would include:
        - Full Gmail API integration
        - Rate limiting and retry logic
        - Comprehensive error handling
        - Pagination support
        - Message parsing and formatting
        """
        import aiohttp
        
        access_token = credentials['access_token']
        limit = parameters.get('limit', 10)
        query = parameters.get('query', 'in:inbox')
        
        headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }
        
        async with aiohttp.ClientSession() as session:
            # List messages
            list_url = f"https://www.googleapis.com/gmail/v1/users/me/messages"
            list_params = {'q': query, 'maxResults': limit}
            
            async with session.get(list_url, headers=headers, params=list_params) as response:
                if response.status == 200:
                    list_data = await response.json()
                    message_ids = [msg['id'] for msg in list_data.get('messages', [])]
                    
                    # Fetch message details
                    messages = []
                    for msg_id in message_ids[:limit]:
                        msg_url = f"https://www.googleapis.com/gmail/v1/users/me/messages/{msg_id}"
                        async with session.get(msg_url, headers=headers) as msg_response:
                            if msg_response.status == 200:
                                msg_data = await msg_response.json()
                                messages.append(self._parse_gmail_message(msg_data))
                    
                    return {
                        'success': True,
                        'data': messages,
                        'count': len(messages),
                        'api_calls': 1 + len(message_ids),
                        'operation': 'read_emails',
                        'provider': 'google'
                    }
                else:
                    raise Exception(f"Gmail API error: {response.status}")

    def _parse_gmail_message(self, msg_data: Dict) -> Dict:
        """Parse Gmail API message response into standardized format"""
        payload = msg_data.get('payload', {})
        headers = {h['name']: h['value'] for h in payload.get('headers', [])}
        
        return {
            'id': msg_data['id'],
            'subject': headers.get('Subject', 'No Subject'),
            'from': headers.get('From', 'Unknown Sender'),
            'to': headers.get('To', ''),
            'date': headers.get('Date', ''),
            'snippet': msg_data.get('snippet', ''),
            'labels': msg_data.get('labelIds', []),
            'thread_id': msg_data.get('threadId', ''),
            'size_estimate': msg_data.get('sizeEstimate', 0)
        }

    # Additional helper methods for other providers and operations...
    async def _fetch_outlook_emails(self, credentials: Dict, parameters: Dict) -> Dict:
        """Microsoft Graph API implementation"""
        # Implementation for Microsoft Graph API
        pass
        
    async def _fetch_yahoo_emails(self, credentials: Dict, parameters: Dict) -> Dict:
        """Yahoo Mail API implementation"""
        # Implementation for Yahoo Mail API
        pass

App-Scoped User Authentication Flow

Users authenticate with third-party services through an app-specific OAuth flow:

1. App Setup - Developer creates OAuth authenticator for their app via API
2. User Initiate - User clicks "Connect Gmail" in the application
3. OAuth Redirect - Application redirects to provider's authorization page
4. User Consent - User grants permissions to the specific application
5. Authorization Callback - Provider redirects back with authorization code
6. Token Exchange - FiberWise exchanges code for access/refresh tokens
7. Secure Storage - Tokens stored with user and app isolation
8. Agent Access - Agents access APIs through BaseCredentialService

OAuth Authenticators API

Manage OAuth authenticators for your applications via the REST API:

🔗 Create OAuth Authenticator

curl -X POST /api/v1/oauth/authenticators/{app_id} \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-api-token" \
  -d '{
    "name": "Gmail OAuth",
    "authenticator_type": "google", 
    "client_id": "your-client-id",
    "client_secret": "your-client-secret",
    "scopes": ["https://www.googleapis.com/auth/gmail.readonly"],
    "authorize_url": "https://accounts.google.com/o/oauth2/auth",
    "token_url": "https://oauth2.googleapis.com/token",
    "redirect_uri": "https://your-app.com/auth/callback/google"
  }'

📋 List App Authenticators

curl -X GET /api/v1/oauth/authenticators?app_id={app_id} \
  -H "Authorization: Bearer your-api-token"

🔍 Get Specific Authenticator

curl -X GET /api/v1/oauth/authenticators/{authenticator_id} \
  -H "Authorization: Bearer your-api-token"

🗑️ Delete Authenticator

curl -X DELETE /api/v1/oauth/authenticators/{authenticator_id} \
  -H "Authorization: Bearer your-api-token"

📊 Response Format

{
  "id": "auth_123456789",
  "app_id": "app_987654321", 
  "name": "Gmail OAuth",
  "authenticator_type": "google",
  "scopes": ["https://www.googleapis.com/auth/gmail.readonly"],
  "is_active": true,
  "created_at": "2024-01-15T10:30:00Z",
  "redirect_uri": "https://your-app.com/auth/callback/google"
}

App Manifest Configuration

Configure OAuth authenticators in your app manifest (example from email-agent-app):

# app_manifest.yaml
name: "Email Agent App"
description: "Email management with multi-provider OAuth support"
version: "2.0.0"

# OAuth Provider Configurations
authenticators:
  - name: "Google OAuth"
    authenticator_type: "google"
    description: "Connect to Gmail for email management"
    permissions:
      - "https://www.googleapis.com/auth/gmail.readonly"
      - "https://www.googleapis.com/auth/gmail.send"
      - "https://www.googleapis.com/auth/gmail.modify"
      - "https://www.googleapis.com/auth/userinfo.email"
    is_default: true

  - name: "Microsoft OAuth"
    authenticator_type: "microsoft"
    description: "Connect to Outlook for email management"
    permissions:
      - "https://graph.microsoft.com/mail.read"
      - "https://graph.microsoft.com/mail.send"
    is_default: false

  - name: "Yahoo OAuth"
    authenticator_type: "yahoo"
    description: "Connect to Yahoo Mail"
    permissions:
      - "mail-r"
      - "mail-w"
    is_default: false

# Agent Dependencies
agent:
  name: "EmailAgent"
  dependencies:
    - "BaseCredentialService"
    - "BaseLLMService"
    - "BaseDataService"
  oauth_required: true