CLARISSA Tutorial 10: API Reference¶

Learning Objectives:

• Understand the CLARISSA REST API
• Make API calls programmatically
• Handle authentication and errors
• Build integrations with external tools

Prerequisites: Notebooks 01-09

Estimated Time: 30 minutes

In [ ]:

# Colab Setup — API Keys & Dependencies
import sys, os
IN_COLAB = 'google.colab' in sys.modules

if IN_COLAB:
    from google.colab import userdata
    # Set keys from Colab Secrets (Settings → Secrets → Add)
    try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY')
    except: pass
    try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY')
    except: pass
    try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN')
    except: pass
    # Fallback: manual input
    if not os.environ.get('ANTHROPIC_API_KEY'):
        import getpass
        for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']:
            if not os.environ.get(key):
                try: os.environ[key] = getpass.getpass(f'{key}: ')
                except: pass

print(f'Environment: {"Colab" if IN_COLAB else "Local"}')
for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']:
    print(f'  {k}: {"✓" if os.environ.get(k) else "✗"}')

# Colab Setup — API Keys & Dependencies import sys, os IN_COLAB = 'google.colab' in sys.modules if IN_COLAB: from google.colab import userdata # Set keys from Colab Secrets (Settings → Secrets → Add) try: os.environ['ANTHROPIC_API_KEY'] = userdata.get('ANTHROPIC_API_KEY') except: pass try: os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY') except: pass try: os.environ['GITLAB_TOKEN'] = userdata.get('GITLAB_TOKEN') except: pass # Fallback: manual input if not os.environ.get('ANTHROPIC_API_KEY'): import getpass for key in ['ANTHROPIC_API_KEY', 'OPENAI_API_KEY', 'GITLAB_TOKEN']: if not os.environ.get(key): try: os.environ[key] = getpass.getpass(f'{key}: ') except: pass print(f'Environment: {"Colab" if IN_COLAB else "Local"}') for k in ['ANTHROPIC_API_KEY','OPENAI_API_KEY','GITLAB_TOKEN']: print(f' {k}: {"✓" if os.environ.get(k) else "✗"}')

API Overview¶

CLARISSA provides a REST API for programmatic access:

Endpoint	Method	Description
/api/v1/chat	POST	Send a message, get response
/api/v1/sessions	GET/POST	Manage sessions
/api/v1/deck	GET	Get current deck
/api/v1/simulate	POST	Run simulation
/api/v1/results	GET	Fetch results
/api/v1/health	GET	Health check

In [ ]:

import requests
import json
from typing import Optional, Dict, Any
from dataclasses import dataclass
import os

# API base URL (configurable)
API_BASE = os.environ.get('CLARISSA_API_URL', 'http://localhost:8000')

print(f"CLARISSA API Reference")
print(f"Base URL: {API_BASE}")

import requests import json from typing import Optional, Dict, Any from dataclasses import dataclass import os # API base URL (configurable) API_BASE = os.environ.get('CLARISSA_API_URL', 'http://localhost:8000') print(f"CLARISSA API Reference") print(f"Base URL: {API_BASE}")

Section 1: API Client¶

A Python client for the CLARISSA API.

In [ ]:

@dataclass
class APIResponse:
    """Standardized API response."""
    success: bool
    data: Optional[Dict] = None
    error: Optional[str] = None
    status_code: int = 200

class CLARISSAClient:
    """Python client for CLARISSA REST API."""
    
    def __init__(self, base_url: str = API_BASE, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key or os.environ.get('CLARISSA_API_KEY')
        self.session_id: Optional[str] = None
    
    def _headers(self) -> Dict[str, str]:
        """Build request headers."""
        headers = {'Content-Type': 'application/json'}
        if self.api_key:
            headers['Authorization'] = f'Bearer {self.api_key}'
        if self.session_id:
            headers['X-Session-ID'] = self.session_id
        return headers
    
    def _request(self, method: str, endpoint: str, 
                 data: Optional[Dict] = None) -> APIResponse:
        """Make API request."""
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self._headers(),
                json=data,
                timeout=30
            )
            
            if response.status_code == 200:
                return APIResponse(
                    success=True,
                    data=response.json(),
                    status_code=response.status_code
                )
            else:
                return APIResponse(
                    success=False,
                    error=response.text,
                    status_code=response.status_code
                )
                
        except requests.exceptions.ConnectionError:
            return APIResponse(
                success=False,
                error=f"Cannot connect to {url}",
                status_code=0
            )
        except Exception as e:
            return APIResponse(
                success=False,
                error=str(e),
                status_code=0
            )
    
    # ========== Session Management ==========
    
    def create_session(self) -> APIResponse:
        """Create a new conversation session."""
        response = self._request('POST', '/api/v1/sessions')
        if response.success and response.data:
            self.session_id = response.data.get('session_id')
        return response
    
    def get_session(self) -> APIResponse:
        """Get current session info."""
        return self._request('GET', f'/api/v1/sessions/{self.session_id}')
    
    def delete_session(self) -> APIResponse:
        """Delete current session."""
        response = self._request('DELETE', f'/api/v1/sessions/{self.session_id}')
        if response.success:
            self.session_id = None
        return response
    
    # ========== Chat ==========
    
    def chat(self, message: str) -> APIResponse:
        """Send a chat message."""
        return self._request('POST', '/api/v1/chat', {'message': message})
    
    # ========== Deck Operations ==========
    
    def get_deck(self, format: str = 'text') -> APIResponse:
        """Get the current deck."""
        return self._request('GET', f'/api/v1/deck?format={format}')
    
    def validate_deck(self) -> APIResponse:
        """Validate the current deck."""
        return self._request('POST', '/api/v1/deck/validate')
    
    # ========== Simulation ==========
    
    def run_simulation(self, options: Optional[Dict] = None) -> APIResponse:
        """Start simulation run."""
        return self._request('POST', '/api/v1/simulate', options or {})
    
    def get_simulation_status(self, run_id: str) -> APIResponse:
        """Check simulation status."""
        return self._request('GET', f'/api/v1/simulate/{run_id}/status')
    
    def get_results(self, run_id: str) -> APIResponse:
        """Get simulation results."""
        return self._request('GET', f'/api/v1/results/{run_id}')
    
    # ========== Utilities ==========
    
    def health_check(self) -> APIResponse:
        """Check API health."""
        return self._request('GET', '/api/v1/health')

print("CLARISSA Client defined")

@dataclass class APIResponse: """Standardized API response.""" success: bool data: Optional[Dict] = None error: Optional[str] = None status_code: int = 200 class CLARISSAClient: """Python client for CLARISSA REST API.""" def __init__(self, base_url: str = API_BASE, api_key: Optional[str] = None): self.base_url = base_url.rstrip('/') self.api_key = api_key or os.environ.get('CLARISSA_API_KEY') self.session_id: Optional[str] = None def _headers(self) -> Dict[str, str]: """Build request headers.""" headers = {'Content-Type': 'application/json'} if self.api_key: headers['Authorization'] = f'Bearer {self.api_key}' if self.session_id: headers['X-Session-ID'] = self.session_id return headers def _request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> APIResponse: """Make API request.""" url = f"{self.base_url}{endpoint}" try: response = requests.request( method=method, url=url, headers=self._headers(), json=data, timeout=30 ) if response.status_code == 200: return APIResponse( success=True, data=response.json(), status_code=response.status_code ) else: return APIResponse( success=False, error=response.text, status_code=response.status_code ) except requests.exceptions.ConnectionError: return APIResponse( success=False, error=f"Cannot connect to {url}", status_code=0 ) except Exception as e: return APIResponse( success=False, error=str(e), status_code=0 ) # ========== Session Management ========== def create_session(self) -> APIResponse: """Create a new conversation session.""" response = self._request('POST', '/api/v1/sessions') if response.success and response.data: self.session_id = response.data.get('session_id') return response def get_session(self) -> APIResponse: """Get current session info.""" return self._request('GET', f'/api/v1/sessions/{self.session_id}') def delete_session(self) -> APIResponse: """Delete current session.""" response = self._request('DELETE', f'/api/v1/sessions/{self.session_id}') if response.success: self.session_id = None return response # ========== Chat ========== def chat(self, message: str) -> APIResponse: """Send a chat message.""" return self._request('POST', '/api/v1/chat', {'message': message}) # ========== Deck Operations ========== def get_deck(self, format: str = 'text') -> APIResponse: """Get the current deck.""" return self._request('GET', f'/api/v1/deck?format={format}') def validate_deck(self) -> APIResponse: """Validate the current deck.""" return self._request('POST', '/api/v1/deck/validate') # ========== Simulation ========== def run_simulation(self, options: Optional[Dict] = None) -> APIResponse: """Start simulation run.""" return self._request('POST', '/api/v1/simulate', options or {}) def get_simulation_status(self, run_id: str) -> APIResponse: """Check simulation status.""" return self._request('GET', f'/api/v1/simulate/{run_id}/status') def get_results(self, run_id: str) -> APIResponse: """Get simulation results.""" return self._request('GET', f'/api/v1/results/{run_id}') # ========== Utilities ========== def health_check(self) -> APIResponse: """Check API health.""" return self._request('GET', '/api/v1/health') print("CLARISSA Client defined")

Section 2: API Endpoints Reference¶

Detailed documentation for each endpoint.

In [ ]:

API_DOCS = {
    'POST /api/v1/chat': {
        'description': 'Send a message to CLARISSA and receive a response',
        'request': {
            'message': 'string (required) - The user message',
            'context': 'object (optional) - Additional context'
        },
        'response': {
            'response': 'string - CLARISSA response',
            'intent': 'string - Detected intent',
            'entities': 'object - Extracted entities',
            'deck_updated': 'boolean - Whether deck changed'
        },
        'example_request': {
            'message': 'Create a 20x20x5 waterflood model'
        },
        'example_response': {
            'response': "I've created a 20x20x5 model...",
            'intent': 'CREATE_MODEL',
            'entities': {'grid_size': [20, 20, 5]},
            'deck_updated': True
        }
    },
    
    'POST /api/v1/sessions': {
        'description': 'Create a new conversation session',
        'request': {
            'name': 'string (optional) - Session name',
            'metadata': 'object (optional) - Custom metadata'
        },
        'response': {
            'session_id': 'string - Unique session ID',
            'created_at': 'string - ISO timestamp'
        }
    },
    
    'GET /api/v1/deck': {
        'description': 'Get the current simulation deck',
        'query_params': {
            'format': 'text|json (default: text)'
        },
        'response': {
            'deck': 'string - The ECLIPSE deck content',
            'sections': 'array - List of sections present',
            'assumptions': 'array - Assumptions made'
        }
    },
    
    'POST /api/v1/deck/validate': {
        'description': 'Validate the current deck',
        'response': {
            'valid': 'boolean - Overall validity',
            'syntactic': 'object - Syntactic validation results',
            'semantic': 'object - Semantic validation results',
            'physics': 'object - Physics validation results'
        }
    },
    
    'POST /api/v1/simulate': {
        'description': 'Start a simulation run',
        'request': {
            'simulator': 'string (default: opm-flow)',
            'parallel': 'integer (optional) - Number of processes',
            'timeout': 'integer (optional) - Max seconds'
        },
        'response': {
            'run_id': 'string - Unique run ID',
            'status': 'string - queued|running|completed|failed'
        }
    },
    
    'GET /api/v1/results/{run_id}': {
        'description': 'Get simulation results',
        'response': {
            'status': 'string - Completion status',
            'summary': 'object - Summary statistics',
            'well_data': 'array - Well production data',
            'field_data': 'object - Field totals'
        }
    },
    
    'GET /api/v1/health': {
        'description': 'Check API and component health',
        'response': {
            'status': 'string - healthy|degraded|unhealthy',
            'components': 'object - Status of each component',
            'version': 'string - API version'
        }
    }
}

# Display API docs
print("CLARISSA API Reference")
print("=" * 70)

for endpoint, doc in API_DOCS.items():
    print(f"\n{endpoint}")
    print("-" * 70)
    print(f"  {doc['description']}")
    
    if 'request' in doc:
        print("\n  Request Body:")
        for param, desc in doc['request'].items():
            print(f"    {param}: {desc}")
    
    if 'query_params' in doc:
        print("\n  Query Parameters:")
        for param, desc in doc['query_params'].items():
            print(f"    {param}: {desc}")

API_DOCS = { 'POST /api/v1/chat': { 'description': 'Send a message to CLARISSA and receive a response', 'request': { 'message': 'string (required) - The user message', 'context': 'object (optional) - Additional context' }, 'response': { 'response': 'string - CLARISSA response', 'intent': 'string - Detected intent', 'entities': 'object - Extracted entities', 'deck_updated': 'boolean - Whether deck changed' }, 'example_request': { 'message': 'Create a 20x20x5 waterflood model' }, 'example_response': { 'response': "I've created a 20x20x5 model...", 'intent': 'CREATE_MODEL', 'entities': {'grid_size': [20, 20, 5]}, 'deck_updated': True } }, 'POST /api/v1/sessions': { 'description': 'Create a new conversation session', 'request': { 'name': 'string (optional) - Session name', 'metadata': 'object (optional) - Custom metadata' }, 'response': { 'session_id': 'string - Unique session ID', 'created_at': 'string - ISO timestamp' } }, 'GET /api/v1/deck': { 'description': 'Get the current simulation deck', 'query_params': { 'format': 'text|json (default: text)' }, 'response': { 'deck': 'string - The ECLIPSE deck content', 'sections': 'array - List of sections present', 'assumptions': 'array - Assumptions made' } }, 'POST /api/v1/deck/validate': { 'description': 'Validate the current deck', 'response': { 'valid': 'boolean - Overall validity', 'syntactic': 'object - Syntactic validation results', 'semantic': 'object - Semantic validation results', 'physics': 'object - Physics validation results' } }, 'POST /api/v1/simulate': { 'description': 'Start a simulation run', 'request': { 'simulator': 'string (default: opm-flow)', 'parallel': 'integer (optional) - Number of processes', 'timeout': 'integer (optional) - Max seconds' }, 'response': { 'run_id': 'string - Unique run ID', 'status': 'string - queued|running|completed|failed' } }, 'GET /api/v1/results/{run_id}': { 'description': 'Get simulation results', 'response': { 'status': 'string - Completion status', 'summary': 'object - Summary statistics', 'well_data': 'array - Well production data', 'field_data': 'object - Field totals' } }, 'GET /api/v1/health': { 'description': 'Check API and component health', 'response': { 'status': 'string - healthy|degraded|unhealthy', 'components': 'object - Status of each component', 'version': 'string - API version' } } } # Display API docs print("CLARISSA API Reference") print("=" * 70) for endpoint, doc in API_DOCS.items(): print(f"\n{endpoint}") print("-" * 70) print(f" {doc['description']}") if 'request' in doc: print("\n Request Body:") for param, desc in doc['request'].items(): print(f" {param}: {desc}") if 'query_params' in doc: print("\n Query Parameters:") for param, desc in doc['query_params'].items(): print(f" {param}: {desc}")

Section 3: Usage Examples¶

Common API usage patterns.

In [ ]:

# Example: Complete workflow using the client

def example_workflow():
    """
    Example: Create model and run simulation via API.
    
    Note: This requires a running CLARISSA server.
    Run 'clarissa serve' to start the API server.
    """
    
    client = CLARISSAClient()
    
    # 1. Check health
    print("1. Checking API health...")
    health = client.health_check()
    if not health.success:
        print(f"   API not available: {health.error}")
        return None
    print(f"   Status: {health.data.get('status', 'unknown')}")
    
    # 2. Create session
    print("\n2. Creating session...")
    session = client.create_session()
    if not session.success:
        print(f"   Failed: {session.error}")
        return None
    print(f"   Session ID: {client.session_id}")
    
    # 3. Build model through conversation
    print("\n3. Building model...")
    messages = [
        "Create a 15x15x4 waterflood model",
        "Set porosity to 0.20 and permeability to 100 md",
        "Add a producer at the center"
    ]
    
    for msg in messages:
        print(f"   User: {msg}")
        response = client.chat(msg)
        if response.success:
            print(f"   CLARISSA: {response.data.get('response', '')[:100]}...")
        else:
            print(f"   Error: {response.error}")
    
    # 4. Validate deck
    print("\n4. Validating deck...")
    validation = client.validate_deck()
    if validation.success:
        print(f"   Valid: {validation.data.get('valid', False)}")
    
    # 5. Run simulation
    print("\n5. Running simulation...")
    sim = client.run_simulation({'timeout': 300})
    if sim.success:
        run_id = sim.data.get('run_id')
        print(f"   Run ID: {run_id}")
        
        # Poll for completion
        import time
        for _ in range(10):
            status = client.get_simulation_status(run_id)
            if status.success:
                state = status.data.get('status')
                print(f"   Status: {state}")
                if state in ['completed', 'failed']:
                    break
            time.sleep(2)
    
    # 6. Get results
    print("\n6. Fetching results...")
    results = client.get_results(run_id)
    if results.success:
        print(f"   Summary: {results.data.get('summary', {})}")
    
    return client

# Run example (will fail without server, but shows the pattern)
print("Example API Workflow")
print("=" * 60)
print("Note: Requires CLARISSA server running at", API_BASE)
print()

example_workflow()

# Example: Complete workflow using the client def example_workflow(): """ Example: Create model and run simulation via API. Note: This requires a running CLARISSA server. Run 'clarissa serve' to start the API server. """ client = CLARISSAClient() # 1. Check health print("1. Checking API health...") health = client.health_check() if not health.success: print(f" API not available: {health.error}") return None print(f" Status: {health.data.get('status', 'unknown')}") # 2. Create session print("\n2. Creating session...") session = client.create_session() if not session.success: print(f" Failed: {session.error}") return None print(f" Session ID: {client.session_id}") # 3. Build model through conversation print("\n3. Building model...") messages = [ "Create a 15x15x4 waterflood model", "Set porosity to 0.20 and permeability to 100 md", "Add a producer at the center" ] for msg in messages: print(f" User: {msg}") response = client.chat(msg) if response.success: print(f" CLARISSA: {response.data.get('response', '')[:100]}...") else: print(f" Error: {response.error}") # 4. Validate deck print("\n4. Validating deck...") validation = client.validate_deck() if validation.success: print(f" Valid: {validation.data.get('valid', False)}") # 5. Run simulation print("\n5. Running simulation...") sim = client.run_simulation({'timeout': 300}) if sim.success: run_id = sim.data.get('run_id') print(f" Run ID: {run_id}") # Poll for completion import time for _ in range(10): status = client.get_simulation_status(run_id) if status.success: state = status.data.get('status') print(f" Status: {state}") if state in ['completed', 'failed']: break time.sleep(2) # 6. Get results print("\n6. Fetching results...") results = client.get_results(run_id) if results.success: print(f" Summary: {results.data.get('summary', {})}") return client # Run example (will fail without server, but shows the pattern) print("Example API Workflow") print("=" * 60) print("Note: Requires CLARISSA server running at", API_BASE) print() example_workflow()

Section 4: cURL Examples¶

For command-line usage and testing.

In [ ]:

CURL_EXAMPLES = '''
# CLARISSA API - cURL Examples
# =============================

# Health check
curl -X GET http://localhost:8000/api/v1/health

# Create session
curl -X POST http://localhost:8000/api/v1/sessions \
  -H "Content-Type: application/json" \
  -d '{"name": "my-session"}'

# Send chat message
curl -X POST http://localhost:8000/api/v1/chat \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: YOUR_SESSION_ID" \
  -d '{"message": "Create a waterflood model with 20x20x5 grid"}'

# Get deck
curl -X GET "http://localhost:8000/api/v1/deck?format=text" \
  -H "X-Session-ID: YOUR_SESSION_ID"

# Validate deck
curl -X POST http://localhost:8000/api/v1/deck/validate \
  -H "X-Session-ID: YOUR_SESSION_ID"

# Run simulation
curl -X POST http://localhost:8000/api/v1/simulate \
  -H "Content-Type: application/json" \
  -H "X-Session-ID: YOUR_SESSION_ID" \
  -d '{"simulator": "opm-flow", "timeout": 300}'

# Get results
curl -X GET http://localhost:8000/api/v1/results/YOUR_RUN_ID \
  -H "X-Session-ID: YOUR_SESSION_ID"
'''

print(CURL_EXAMPLES)

CURL_EXAMPLES = ''' # CLARISSA API - cURL Examples # ============================= # Health check curl -X GET http://localhost:8000/api/v1/health # Create session curl -X POST http://localhost:8000/api/v1/sessions \ -H "Content-Type: application/json" \ -d '{"name": "my-session"}' # Send chat message curl -X POST http://localhost:8000/api/v1/chat \ -H "Content-Type: application/json" \ -H "X-Session-ID: YOUR_SESSION_ID" \ -d '{"message": "Create a waterflood model with 20x20x5 grid"}' # Get deck curl -X GET "http://localhost:8000/api/v1/deck?format=text" \ -H "X-Session-ID: YOUR_SESSION_ID" # Validate deck curl -X POST http://localhost:8000/api/v1/deck/validate \ -H "X-Session-ID: YOUR_SESSION_ID" # Run simulation curl -X POST http://localhost:8000/api/v1/simulate \ -H "Content-Type: application/json" \ -H "X-Session-ID: YOUR_SESSION_ID" \ -d '{"simulator": "opm-flow", "timeout": 300}' # Get results curl -X GET http://localhost:8000/api/v1/results/YOUR_RUN_ID \ -H "X-Session-ID: YOUR_SESSION_ID" ''' print(CURL_EXAMPLES)

Section 5: Error Handling¶

Common errors and how to handle them.

In [ ]:

ERROR_CODES = {
    400: {
        'name': 'Bad Request',
        'description': 'Invalid request format or missing required fields',
        'example': 'Missing "message" field in chat request',
        'action': 'Check request body matches API spec'
    },
    401: {
        'name': 'Unauthorized',
        'description': 'Invalid or missing API key',
        'example': 'Authorization header missing or invalid',
        'action': 'Include valid Bearer token in Authorization header'
    },
    404: {
        'name': 'Not Found',
        'description': 'Requested resource does not exist',
        'example': 'Session ID not found',
        'action': 'Verify resource ID is correct, create if needed'
    },
    422: {
        'name': 'Unprocessable Entity',
        'description': 'Request understood but cannot be processed',
        'example': 'Deck validation failed',
        'action': 'Check deck content and fix validation errors'
    },
    429: {
        'name': 'Rate Limited',
        'description': 'Too many requests',
        'example': 'Exceeded 100 requests/minute',
        'action': 'Implement backoff, wait before retrying'
    },
    500: {
        'name': 'Internal Server Error',
        'description': 'Server-side error',
        'example': 'Simulation engine crashed',
        'action': 'Retry request, contact support if persistent'
    },
    503: {
        'name': 'Service Unavailable',
        'description': 'Service temporarily down',
        'example': 'Simulation queue full',
        'action': 'Wait and retry with exponential backoff'
    }
}

print("Error Code Reference")
print("=" * 60)

for code, info in ERROR_CODES.items():
    print(f"\n{code} - {info['name']}")
    print(f"  Description: {info['description']}")
    print(f"  Example: {info['example']}")
    print(f"  Action: {info['action']}")

ERROR_CODES = { 400: { 'name': 'Bad Request', 'description': 'Invalid request format or missing required fields', 'example': 'Missing "message" field in chat request', 'action': 'Check request body matches API spec' }, 401: { 'name': 'Unauthorized', 'description': 'Invalid or missing API key', 'example': 'Authorization header missing or invalid', 'action': 'Include valid Bearer token in Authorization header' }, 404: { 'name': 'Not Found', 'description': 'Requested resource does not exist', 'example': 'Session ID not found', 'action': 'Verify resource ID is correct, create if needed' }, 422: { 'name': 'Unprocessable Entity', 'description': 'Request understood but cannot be processed', 'example': 'Deck validation failed', 'action': 'Check deck content and fix validation errors' }, 429: { 'name': 'Rate Limited', 'description': 'Too many requests', 'example': 'Exceeded 100 requests/minute', 'action': 'Implement backoff, wait before retrying' }, 500: { 'name': 'Internal Server Error', 'description': 'Server-side error', 'example': 'Simulation engine crashed', 'action': 'Retry request, contact support if persistent' }, 503: { 'name': 'Service Unavailable', 'description': 'Service temporarily down', 'example': 'Simulation queue full', 'action': 'Wait and retry with exponential backoff' } } print("Error Code Reference") print("=" * 60) for code, info in ERROR_CODES.items(): print(f"\n{code} - {info['name']}") print(f" Description: {info['description']}") print(f" Example: {info['example']}") print(f" Action: {info['action']}")

In [ ]:

# Robust client with retry logic

import time

class RobustCLARISSAClient(CLARISSAClient):
    """Client with retry and error handling."""
    
    def __init__(self, *args, max_retries: int = 3, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_retries = max_retries
    
    def _request_with_retry(self, method: str, endpoint: str,
                            data: Optional[Dict] = None) -> APIResponse:
        """Make request with exponential backoff retry."""
        last_response = None
        
        for attempt in range(self.max_retries):
            response = self._request(method, endpoint, data)
            last_response = response
            
            # Success
            if response.success:
                return response
            
            # Don't retry client errors (4xx)
            if 400 <= response.status_code < 500:
                return response
            
            # Retry server errors (5xx) and network errors
            if response.status_code >= 500 or response.status_code == 0:
                wait_time = (2 ** attempt) + (time.time() % 1)
                print(f"  Retry {attempt + 1}/{self.max_retries} in {wait_time:.1f}s...")
                time.sleep(wait_time)
        
        return last_response
    
    def chat(self, message: str) -> APIResponse:
        """Chat with retry."""
        return self._request_with_retry('POST', '/api/v1/chat', {'message': message})

print("RobustCLARISSAClient defined with retry logic")

# Robust client with retry logic import time class RobustCLARISSAClient(CLARISSAClient): """Client with retry and error handling.""" def __init__(self, *args, max_retries: int = 3, **kwargs): super().__init__(*args, **kwargs) self.max_retries = max_retries def _request_with_retry(self, method: str, endpoint: str, data: Optional[Dict] = None) -> APIResponse: """Make request with exponential backoff retry.""" last_response = None for attempt in range(self.max_retries): response = self._request(method, endpoint, data) last_response = response # Success if response.success: return response # Don't retry client errors (4xx) if 400 <= response.status_code < 500: return response # Retry server errors (5xx) and network errors if response.status_code >= 500 or response.status_code == 0: wait_time = (2 ** attempt) + (time.time() % 1) print(f" Retry {attempt + 1}/{self.max_retries} in {wait_time:.1f}s...") time.sleep(wait_time) return last_response def chat(self, message: str) -> APIResponse: """Chat with retry.""" return self._request_with_retry('POST', '/api/v1/chat', {'message': message}) print("RobustCLARISSAClient defined with retry logic")

Summary¶

In this tutorial, we learned:

1. API Overview: RESTful endpoints for CLARISSA
2. Python Client: Programmatic access with CLARISSAClient
3. Endpoint Reference: Detailed docs for each endpoint
4. cURL Examples: Command-line testing
5. Error Handling: Status codes and retry logic

Key Insight: The API enables integration with external tools, CI/CD pipelines, and custom workflows.

---

Congratulations!¶

You've completed the CLARISSA Tutorial Series. You now understand:

• ECLIPSE deck fundamentals
• OPM Flow integration
• Knowledge layer and RAG
• LLM conversation management
• Physics constraints with Z3
• Template-based deck generation
• Reinforcement learning for optimization
• RIGOR benchmark evaluation
• End-to-end pipeline
• REST API usage

Next Steps:

• Deploy CLARISSA in your environment
• Contribute to the project on GitLab
• Build custom integrations