Error Handling

Handle API errors with typed exceptions, stream error events, and built-in retry logic.

This guide covers error handling when using the Concurrence SDK in Python or TypeScript. Both SDKs include built-in retry logic for transient failures and provide full error handling for all API operations.

Error Types

Both SDKs provide typed error handling for different types of failures:

Python

from amigo_sdk.errors import (
    AmigoError,           # Base error class
    AuthenticationError,   # Invalid credentials (401)
    ForbiddenError,       # Permission denied (403)
    NotFoundError,        # Resource not found (404)
    BadRequestError,      # Invalid request (400)
    ValidationError,      # Request validation failed (422)
    ConflictError,        # Resource conflict (409)
    RateLimitError,       # Rate limit (429)
    ServerError,          # Server error (5xx)
    ServiceUnavailableError,  # Service unavailable (503)
)

TypeScript

import {
  AmigoError,           // Base error class
  AuthenticationError,  // Invalid credentials (401)
  PermissionError,      // Permission denied (403)
  NotFoundError,        // Resource not found (404)
  BadRequestError,      // Invalid request (400)
  ValidationError,      // Request validation failed (422)
  ConflictError,        // Resource conflict (409)
  RateLimitError,       // Rate limit (429)
  ServerError,          // Server error (5xx)
  ServiceUnavailableError, // Service unavailable (503)
  NetworkError,         // Network/connection failures
} from '@amigo-ai/sdk'

Both SDKs map HTTP status codes to typed errors, including RateLimitError for 429 responses; any failure without a dedicated class is raised through the base AmigoError. The SDKs differ on connection failures: the TypeScript SDK wraps them in a typed NetworkError, while the Python SDK lets the underlying httpx exceptions (for example httpx.ConnectError and httpx.TimeoutException) propagate.

Basic Error Handling

Python

from amigo_sdk import AmigoClient
from amigo_sdk.errors import (
    AuthenticationError,
    NotFoundError,
    BadRequestError,
    ValidationError,
    ConflictError,
    RateLimitError,
    AmigoError
)

def handle_errors_example():
    try:
        with AmigoClient(
            api_key="your-api-key",
            api_key_id="your-api-key-id",
            user_id="your-user-id",
            organization_id="your-org-id"
        ) as client:
            # Your API calls here
            org = client.organizations.get()
            print(f"Organization: {org.org_name}")
            
    except AuthenticationError as e:
        print(f"Authentication failed: {e}")
        # Handle invalid credentials. Check API keys.
        
    except NotFoundError as e:
        print(f"Resource not found: {e}")
        # Handle missing resources. Check IDs.
        
    except BadRequestError as e:
        print(f"Bad request: {e}")
        # Handle invalid request format
        
    except ValidationError as e:
        print(f"Validation failed: {e}")
        # Handle request validation errors
        
    except ConflictError as e:
        print(f"Conflict: {e}")
        # Handle resource conflicts (often recoverable)
        
    except RateLimitError as e:
        print(f"Rate limited: {e}")
        # Handle 429 rate limiting. GET requests (and POSTs with a
        # Retry-After header) are retried automatically before this is raised.
        
    except AmigoError as e:
        print(f"Concurrence API error: {e}")
        # Handle any other Concurrence-specific errors
        
    except Exception as e:
        print(f"Unexpected error: {e}")
        # Handle unexpected errors

TypeScript

import {
  AmigoClient,
  AmigoError,
  AuthenticationError,
  NotFoundError,
  BadRequestError,
  ValidationError,
  ConflictError,
  RateLimitError,
  NetworkError,
} from '@amigo-ai/sdk'

async function handleErrorsExample(): Promise<void> {
  const client = new AmigoClient({
    apiKey: 'your-api-key',
    apiKeyId: 'your-api-key-id',
    userId: 'your-user-id',
    orgId: 'your-org-id'
  })

  try {
    // Your API calls here
    const org = await client.organizations.getOrganization()
    console.log(`Organization: ${org.org_name}`)
    
  } catch (error) {
    if (error instanceof AuthenticationError) {
      console.log(`Authentication failed: ${error.message}`)
      // Handle invalid credentials. Check API keys.
      
    } else if (error instanceof NotFoundError) {
      console.log(`Resource not found: ${error.message}`)
      // Handle missing resources. Check IDs.
      
    } else if (error instanceof BadRequestError) {
      console.log(`Bad request: ${error.message}`)
      // Handle invalid request format
      
    } else if (error instanceof ValidationError) {
      console.log(`Validation failed: ${error.message}`)
      // Handle request validation errors
      
    } else if (error instanceof ConflictError) {
      console.log(`Conflict: ${error.message}`)
      // Handle resource conflicts (often recoverable)
      
    } else if (error instanceof RateLimitError) {
      console.log(`Rate limited: ${error.message}`)
      // Handle 429 rate limiting. GET requests (and POSTs with a
      // Retry-After header) are retried automatically before this is thrown.
      
    } else if (error instanceof NetworkError) {
      console.log(`Network error: ${error.message}`)
      // Handle connection failures. GET requests are retried automatically.
      
    } else if (error instanceof AmigoError) {
      console.log(`Concurrence API error: ${error.message}`)
      // Handle any other Concurrence-specific errors
      
    } else {
      console.log(`Unexpected error:`, error)
      // Handle unexpected errors
    }
  }
}

Stream Error Handling

When processing streaming responses, handle errors within the stream. See also Conversations: Interact.

Python

from amigo_sdk import AmigoClient
from amigo_sdk.models import (
    ConversationCreateConversationRequest,
    CreateConversationParametersQuery,
)

def handle_stream_errors():
    with AmigoClient() as client:
        try:
            events = client.conversations.create_conversation(
                ConversationCreateConversationRequest(service_id="your-service-id"),
                CreateConversationParametersQuery(response_format="text"),
            )
            
            for event in events:
                event_data = event.model_dump(mode="json")
                
                # Check for error events in the stream
                if event_data.get("type") == "error":
                    description = event_data.get("error_description", "Unknown error")
                    status = event_data.get("http_error_code")
                    print(f"Stream error ({status}): {description}")
                    break
                    
                # Process normal events
                print(f"Event: {event_data.get('type')}")
                
        except Exception as e:
            print(f"Stream processing error: {e}")

TypeScript

async function handleStreamErrors(): Promise<void> {
  const client = new AmigoClient({ ...config })

  try {
    const events = await client.conversations.createConversation({
      body: { service_id: 'your-service-id' },
      query: { response_format: 'text' },
    })

    for await (const event of events) {
      // Check for error events in the stream
      if (event.type === 'error') {
        console.log(`Stream error (${event.http_error_code}): ${event.error_description}`)
        break
      }

      // Process normal events
      console.log(`Event: ${event.type}`)
    }
  } catch (error) {
    console.log(`Stream processing error:`, error)
  }
}

Built-in Retry Logic

Both SDKs share the same default retry policy for transient failures:

Setting	Default
Max attempts	3
Backoff	Exponential with full jitter, 250ms base, 30s max delay per attempt
Retried statuses	408, 429, 500, 502, 503, 504
Retried methods	GET, plus POST when the response is a 429 with a Retry-After header
Retry-After header	Honored (numeric or HTTP-date), capped at the max delay

GET requests are retried; POST requests generally are not. By default, only GET requests are retried on network errors and retryable statuses. POST requests (including conversation creation and interactions) are retried only for a 429 response that carries a Retry-After header; other POST failures surface immediately so you can decide whether to resend.

You can tune this policy: the TypeScript client accepts a retry option (maxAttempts, backoffBaseMs, maxDelayMs, retryOnStatus, retryOnMethods), and the Python client accepts retry_max_attempts, retry_backoff_base, retry_max_delay_seconds, retry_on_status, and retry_on_methods keyword arguments.

Retry Flow Diagram

Best Practices

1. Handle specific error types. Don't just catch generic exceptions.
2. Log errors with context. Include relevant IDs and operation details.
3. Check for stream error events. Monitor the error event type in streams.
4. Trust built-in retries. Let the SDK retry transient GET failures automatically; handle failed POST requests explicitly.
5. Graceful degradation. Provide fallback behavior when possible.
6. Monitor error patterns. Track error rates and types in production.

Next Steps

With proper error handling in place, you're ready to:

• Work with conversations. Streaming and reliable handlers.
• Manage users. User management with error handling.