CLARISSA Tutorial 11: CRUD Playground¶

Interactive playground for CLARISSA API operations

This notebook provides a hands-on environment to explore Create, Read, Update, and Delete operations against the CLARISSA API. Use it to:

• 🧪 Test API endpoints interactively
• 📝 Create and manage simulation sessions
• 🔧 Modify reservoir models via conversation
• 📊 Retrieve and analyze simulation results

No prior CLARISSA experience required!

---

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 "✗"}')

Setup¶

First, let's configure the environment and connect to the CLARISSA API.

In [ ]:

# Configuration
import os
import json
import requests
from datetime import datetime
from typing import Optional, Dict, Any

# API Configuration (adjust for your deployment)
API_BASE_URL = os.getenv('CLARISSA_API_URL', 'http://localhost:8000/api/v1')
API_KEY = os.getenv('CLARISSA_API_KEY', 'demo-key')  # For authenticated endpoints

print(f"🔗 CLARISSA API: {API_BASE_URL}")
print(f"📅 Session started: {datetime.now().isoformat()}")

# Configuration import os import json import requests from datetime import datetime from typing import Optional, Dict, Any # API Configuration (adjust for your deployment) API_BASE_URL = os.getenv('CLARISSA_API_URL', 'http://localhost:8000/api/v1') API_KEY = os.getenv('CLARISSA_API_KEY', 'demo-key') # For authenticated endpoints print(f"🔗 CLARISSA API: {API_BASE_URL}") print(f"📅 Session started: {datetime.now().isoformat()}")

In [ ]:

# Helper class for API interactions
class ClarissaClient:
    """Simple client for CLARISSA CRUD operations."""
    
    def __init__(self, base_url: str, api_key: str = None):
        self.base_url = base_url.rstrip('/')
        self.headers = {'Content-Type': 'application/json'}
        if api_key:
            self.headers['Authorization'] = f'Bearer {api_key}'
        self.session_id = None
    
    def _request(self, method: str, endpoint: str, data: Dict = None) -> Dict:
        """Make API request with error handling."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                json=data,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.ConnectionError:
            return {'error': f'Cannot connect to {url}', 'status': 'offline'}
        except requests.exceptions.HTTPError as e:
            return {'error': str(e), 'status_code': response.status_code}
        except Exception as e:
            return {'error': str(e)}
    
    # Health check
    def health(self) -> Dict:
        return self._request('GET', '/health')
    
    # === CREATE Operations ===
    def create_session(self, name: str = None, model_type: str = 'blackoil') -> Dict:
        """Create a new conversation session."""
        data = {
            'name': name or f'Session_{datetime.now().strftime("%Y%m%d_%H%M%S")}',
            'model_type': model_type
        }
        result = self._request('POST', '/sessions', data)
        if 'session_id' in result:
            self.session_id = result['session_id']
        return result
    
    def send_message(self, message: str, session_id: str = None) -> Dict:
        """Send a message to CLARISSA."""
        sid = session_id or self.session_id
        if not sid:
            return {'error': 'No session. Call create_session() first.'}
        data = {'message': message, 'session_id': sid}
        return self._request('POST', '/chat', data)
    
    def run_simulation(self, session_id: str = None) -> Dict:
        """Execute OPM Flow simulation."""
        sid = session_id or self.session_id
        data = {'session_id': sid}
        return self._request('POST', '/simulate', data)
    
    # === READ Operations ===
    def list_sessions(self) -> Dict:
        """List all sessions."""
        return self._request('GET', '/sessions')
    
    def get_session(self, session_id: str = None) -> Dict:
        """Get session details."""
        sid = session_id or self.session_id
        return self._request('GET', f'/sessions/{sid}')
    
    def get_deck(self, session_id: str = None) -> Dict:
        """Get current ECLIPSE deck."""
        sid = session_id or self.session_id
        return self._request('GET', f'/deck?session_id={sid}')
    
    def get_results(self, session_id: str = None) -> Dict:
        """Get simulation results."""
        sid = session_id or self.session_id
        return self._request('GET', f'/results?session_id={sid}')
    
    def get_conversation(self, session_id: str = None) -> Dict:
        """Get conversation history."""
        sid = session_id or self.session_id
        return self._request('GET', f'/sessions/{sid}/conversation')
    
    # === UPDATE Operations ===
    def update_session(self, session_id: str = None, name: str = None, metadata: Dict = None) -> Dict:
        """Update session properties."""
        sid = session_id or self.session_id
        data = {}
        if name:
            data['name'] = name
        if metadata:
            data['metadata'] = metadata
        return self._request('PATCH', f'/sessions/{sid}', data)
    
    def update_deck(self, deck_content: str, session_id: str = None) -> Dict:
        """Directly update deck content."""
        sid = session_id or self.session_id
        data = {'deck': deck_content, 'session_id': sid}
        return self._request('PUT', '/deck', data)
    
    # === DELETE Operations ===
    def delete_session(self, session_id: str) -> Dict:
        """Delete a session."""
        return self._request('DELETE', f'/sessions/{session_id}')
    
    def clear_conversation(self, session_id: str = None) -> Dict:
        """Clear conversation history but keep session."""
        sid = session_id or self.session_id
        return self._request('DELETE', f'/sessions/{sid}/conversation')

# Initialize client
client = ClarissaClient(API_BASE_URL, API_KEY)
print("✅ ClarissaClient initialized")

# Helper class for API interactions class ClarissaClient: """Simple client for CLARISSA CRUD operations.""" def __init__(self, base_url: str, api_key: str = None): self.base_url = base_url.rstrip('/') self.headers = {'Content-Type': 'application/json'} if api_key: self.headers['Authorization'] = f'Bearer {api_key}' self.session_id = None def _request(self, method: str, endpoint: str, data: Dict = None) -> Dict: """Make API request with error handling.""" url = f"{self.base_url}/{endpoint.lstrip('/')}" try: response = requests.request( method=method, url=url, headers=self.headers, json=data, timeout=30 ) response.raise_for_status() return response.json() except requests.exceptions.ConnectionError: return {'error': f'Cannot connect to {url}', 'status': 'offline'} except requests.exceptions.HTTPError as e: return {'error': str(e), 'status_code': response.status_code} except Exception as e: return {'error': str(e)} # Health check def health(self) -> Dict: return self._request('GET', '/health') # === CREATE Operations === def create_session(self, name: str = None, model_type: str = 'blackoil') -> Dict: """Create a new conversation session.""" data = { 'name': name or f'Session_{datetime.now().strftime("%Y%m%d_%H%M%S")}', 'model_type': model_type } result = self._request('POST', '/sessions', data) if 'session_id' in result: self.session_id = result['session_id'] return result def send_message(self, message: str, session_id: str = None) -> Dict: """Send a message to CLARISSA.""" sid = session_id or self.session_id if not sid: return {'error': 'No session. Call create_session() first.'} data = {'message': message, 'session_id': sid} return self._request('POST', '/chat', data) def run_simulation(self, session_id: str = None) -> Dict: """Execute OPM Flow simulation.""" sid = session_id or self.session_id data = {'session_id': sid} return self._request('POST', '/simulate', data) # === READ Operations === def list_sessions(self) -> Dict: """List all sessions.""" return self._request('GET', '/sessions') def get_session(self, session_id: str = None) -> Dict: """Get session details.""" sid = session_id or self.session_id return self._request('GET', f'/sessions/{sid}') def get_deck(self, session_id: str = None) -> Dict: """Get current ECLIPSE deck.""" sid = session_id or self.session_id return self._request('GET', f'/deck?session_id={sid}') def get_results(self, session_id: str = None) -> Dict: """Get simulation results.""" sid = session_id or self.session_id return self._request('GET', f'/results?session_id={sid}') def get_conversation(self, session_id: str = None) -> Dict: """Get conversation history.""" sid = session_id or self.session_id return self._request('GET', f'/sessions/{sid}/conversation') # === UPDATE Operations === def update_session(self, session_id: str = None, name: str = None, metadata: Dict = None) -> Dict: """Update session properties.""" sid = session_id or self.session_id data = {} if name: data['name'] = name if metadata: data['metadata'] = metadata return self._request('PATCH', f'/sessions/{sid}', data) def update_deck(self, deck_content: str, session_id: str = None) -> Dict: """Directly update deck content.""" sid = session_id or self.session_id data = {'deck': deck_content, 'session_id': sid} return self._request('PUT', '/deck', data) # === DELETE Operations === def delete_session(self, session_id: str) -> Dict: """Delete a session.""" return self._request('DELETE', f'/sessions/{session_id}') def clear_conversation(self, session_id: str = None) -> Dict: """Clear conversation history but keep session.""" sid = session_id or self.session_id return self._request('DELETE', f'/sessions/{sid}/conversation') # Initialize client client = ClarissaClient(API_BASE_URL, API_KEY) print("✅ ClarissaClient initialized")

---

🏥 Health Check¶

First, let's verify the API is reachable:

In [ ]:

# Check API health
health = client.health()
print(json.dumps(health, indent=2))

if health.get('status') == 'offline':
    print("
⚠️  API is offline. Running in demo mode with mock responses.")
    print("    To use a real API, set CLARISSA_API_URL environment variable.")

# Check API health health = client.health() print(json.dumps(health, indent=2)) if health.get('status') == 'offline': print(" ⚠️ API is offline. Running in demo mode with mock responses.") print(" To use a real API, set CLARISSA_API_URL environment variable.")

---

🆕 CREATE Operations¶

Create a New Session¶

In [ ]:

# CREATE: New session
session = client.create_session(
    name="CRUD Playground Test",
    model_type="blackoil"
)
print("📝 Created session:")
print(json.dumps(session, indent=2))

# CREATE: New session session = client.create_session( name="CRUD Playground Test", model_type="blackoil" ) print("📝 Created session:") print(json.dumps(session, indent=2))

Send Messages (Create Conversation)¶

In [ ]:

# CREATE: Send messages to build a model
messages = [
    "I need a simple waterflood model",
    "Use a 5-spot pattern with 40 acre spacing",
    "Reservoir depth is 8500 ft, initial pressure 3800 psi"
]

for msg in messages:
    print(f"
👤 User: {msg}")
    response = client.send_message(msg)
    if 'response' in response:
        print(f"🤖 CLARISSA: {response['response'][:200]}...")
    else:
        print(f"   Response: {response}")

# CREATE: Send messages to build a model messages = [ "I need a simple waterflood model", "Use a 5-spot pattern with 40 acre spacing", "Reservoir depth is 8500 ft, initial pressure 3800 psi" ] for msg in messages: print(f" 👤 User: {msg}") response = client.send_message(msg) if 'response' in response: print(f"🤖 CLARISSA: {response['response'][:200]}...") else: print(f" Response: {response}")

Run Simulation (Create Results)¶

In [ ]:

# CREATE: Run simulation to generate results
print("🚀 Starting simulation...")
sim_result = client.run_simulation()
print(json.dumps(sim_result, indent=2))

# CREATE: Run simulation to generate results print("🚀 Starting simulation...") sim_result = client.run_simulation() print(json.dumps(sim_result, indent=2))

---

📖 READ Operations¶

List All Sessions¶

In [ ]:

# READ: List all sessions
sessions = client.list_sessions()
print("📋 All sessions:")
print(json.dumps(sessions, indent=2))

# READ: List all sessions sessions = client.list_sessions() print("📋 All sessions:") print(json.dumps(sessions, indent=2))

Get Session Details¶

In [ ]:

# READ: Get current session details
session_details = client.get_session()
print("📄 Session details:")
print(json.dumps(session_details, indent=2))

# READ: Get current session details session_details = client.get_session() print("📄 Session details:") print(json.dumps(session_details, indent=2))

Get Generated Deck¶

In [ ]:

# READ: Get the generated ECLIPSE deck
deck = client.get_deck()
print("📜 Generated ECLIPSE Deck:")
if 'deck' in deck:
    # Show first 50 lines
    lines = deck['deck'].split('
')[:50]
    for line in lines:
        print(line)
    if len(deck['deck'].split('
')) > 50:
        print(f"
... ({len(deck['deck'].split(chr(10)))} total lines)")
else:
    print(json.dumps(deck, indent=2))

# READ: Get the generated ECLIPSE deck deck = client.get_deck() print("📜 Generated ECLIPSE Deck:") if 'deck' in deck: # Show first 50 lines lines = deck['deck'].split(' ')[:50] for line in lines: print(line) if len(deck['deck'].split(' ')) > 50: print(f" ... ({len(deck['deck'].split(chr(10)))} total lines)") else: print(json.dumps(deck, indent=2))

Get Simulation Results¶

In [ ]:

# READ: Get simulation results
results = client.get_results()
print("📊 Simulation Results:")
print(json.dumps(results, indent=2))

# READ: Get simulation results results = client.get_results() print("📊 Simulation Results:") print(json.dumps(results, indent=2))

Get Conversation History¶

In [ ]:

# READ: Get conversation history
conversation = client.get_conversation()
print("💬 Conversation History:")
if 'messages' in conversation:
    for msg in conversation['messages']:
        role = '👤' if msg['role'] == 'user' else '🤖'
        print(f"{role} {msg['content'][:100]}...")
else:
    print(json.dumps(conversation, indent=2))

# READ: Get conversation history conversation = client.get_conversation() print("💬 Conversation History:") if 'messages' in conversation: for msg in conversation['messages']: role = '👤' if msg['role'] == 'user' else '🤖' print(f"{role} {msg['content'][:100]}...") else: print(json.dumps(conversation, indent=2))

---

✏️ UPDATE Operations¶

Update Session Metadata¶

In [ ]:

# UPDATE: Modify session properties
updated = client.update_session(
    name="CRUD Playground - Updated",
    metadata={
        'project': 'Tutorial',
        'author': 'CLARISSA User',
        'tags': ['waterflood', '5-spot', 'test']
    }
)
print("✏️ Updated session:")
print(json.dumps(updated, indent=2))

# UPDATE: Modify session properties updated = client.update_session( name="CRUD Playground - Updated", metadata={ 'project': 'Tutorial', 'author': 'CLARISSA User', 'tags': ['waterflood', '5-spot', 'test'] } ) print("✏️ Updated session:") print(json.dumps(updated, indent=2))

Update Deck Directly¶

In [ ]:

# UPDATE: Modify deck content directly
# (useful for advanced users who want to tweak the generated deck)

modified_deck = """
-- Modified by CRUD Playground
RUNSPEC
TITLE
  CRUD Playground Demo Model /

DIMENS
  10 10 5 /

OIL
WATER

METRIC

START
  1 'JAN' 2026 /

-- ... rest of deck ...
"""

update_result = client.update_deck(modified_deck)
print("✏️ Deck update result:")
print(json.dumps(update_result, indent=2))

# UPDATE: Modify deck content directly # (useful for advanced users who want to tweak the generated deck) modified_deck = """ -- Modified by CRUD Playground RUNSPEC TITLE CRUD Playground Demo Model / DIMENS 10 10 5 / OIL WATER METRIC START 1 'JAN' 2026 / -- ... rest of deck ... """ update_result = client.update_deck(modified_deck) print("✏️ Deck update result:") print(json.dumps(update_result, indent=2))

Refine Model via Conversation¶

In [ ]:

# UPDATE: Refine the model through conversation
refinements = [
    "Change the permeability to 150 md",
    "Add a gas cap with 50 ft thickness",
    "Increase the simulation time to 20 years"
]

print("🔄 Refining model...
")
for msg in refinements:
    print(f"👤 User: {msg}")
    response = client.send_message(msg)
    if 'response' in response:
        print(f"🤖 CLARISSA: {response['response'][:150]}...
")
    else:
        print(f"   {response}
")

# UPDATE: Refine the model through conversation refinements = [ "Change the permeability to 150 md", "Add a gas cap with 50 ft thickness", "Increase the simulation time to 20 years" ] print("🔄 Refining model... ") for msg in refinements: print(f"👤 User: {msg}") response = client.send_message(msg) if 'response' in response: print(f"🤖 CLARISSA: {response['response'][:150]}... ") else: print(f" {response} ")

---

🗑️ DELETE Operations¶

Clear Conversation History¶

In [ ]:

# DELETE: Clear conversation but keep session
clear_result = client.clear_conversation()
print("🧹 Cleared conversation:")
print(json.dumps(clear_result, indent=2))

# DELETE: Clear conversation but keep session clear_result = client.clear_conversation() print("🧹 Cleared conversation:") print(json.dumps(clear_result, indent=2))

Delete Session¶

⚠️ Warning: This permanently deletes the session and all associated data!

In [ ]:

# DELETE: Remove session entirely
# Uncomment to execute:

# if client.session_id:
#     delete_result = client.delete_session(client.session_id)
#     print("🗑️ Deleted session:")
#     print(json.dumps(delete_result, indent=2))
#     client.session_id = None

print("⚠️ Delete is commented out for safety.")
print("   Uncomment the code above to delete the session.")

# DELETE: Remove session entirely # Uncomment to execute: # if client.session_id: # delete_result = client.delete_session(client.session_id) # print("🗑️ Deleted session:") # print(json.dumps(delete_result, indent=2)) # client.session_id = None print("⚠️ Delete is commented out for safety.") print(" Uncomment the code above to delete the session.")

---

🧪 Interactive Playground¶

Use the cells below to experiment with your own CRUD operations:

In [ ]:

# 🎮 YOUR PLAYGROUND
# Try your own API calls here!

# Example: Create a new session with custom name
# my_session = client.create_session(name="My Custom Model")

# Example: Send a custom message
# response = client.send_message("Build me a gas reservoir model")

# Example: Get all sessions
# all_sessions = client.list_sessions()

print("🎮 Edit this cell to try your own operations!")

# 🎮 YOUR PLAYGROUND # Try your own API calls here! # Example: Create a new session with custom name # my_session = client.create_session(name="My Custom Model") # Example: Send a custom message # response = client.send_message("Build me a gas reservoir model") # Example: Get all sessions # all_sessions = client.list_sessions() print("🎮 Edit this cell to try your own operations!")

In [ ]:

# 📋 Quick Reference
print("""
╔══════════════════════════════════════════════════════════════════╗
║                    CLARISSA CRUD Quick Reference                 ║
╠══════════════════════════════════════════════════════════════════╣
║ CREATE                                                           ║
║   client.create_session(name, model_type)  → New session         ║
║   client.send_message(message)             → Chat response       ║
║   client.run_simulation()                  → Simulation results  ║
╠══════════════════════════════════════════════════════════════════╣
║ READ                                                             ║
║   client.list_sessions()                   → All sessions        ║
║   client.get_session(session_id)           → Session details     ║
║   client.get_deck()                        → ECLIPSE deck        ║
║   client.get_results()                     → Simulation output   ║
║   client.get_conversation()                → Chat history        ║
╠══════════════════════════════════════════════════════════════════╣
║ UPDATE                                                           ║
║   client.update_session(name, metadata)    → Modify session      ║
║   client.update_deck(deck_content)         → Replace deck        ║
║   client.send_message(refinement)          → Refine via chat     ║
╠══════════════════════════════════════════════════════════════════╣
║ DELETE                                                           ║
║   client.clear_conversation()              → Clear history       ║
║   client.delete_session(session_id)        → Remove session      ║
╚══════════════════════════════════════════════════════════════════╝
""")

# 📋 Quick Reference print(""" ╔══════════════════════════════════════════════════════════════════╗ ║ CLARISSA CRUD Quick Reference ║ ╠══════════════════════════════════════════════════════════════════╣ ║ CREATE ║ ║ client.create_session(name, model_type) → New session ║ ║ client.send_message(message) → Chat response ║ ║ client.run_simulation() → Simulation results ║ ╠══════════════════════════════════════════════════════════════════╣ ║ READ ║ ║ client.list_sessions() → All sessions ║ ║ client.get_session(session_id) → Session details ║ ║ client.get_deck() → ECLIPSE deck ║ ║ client.get_results() → Simulation output ║ ║ client.get_conversation() → Chat history ║ ╠══════════════════════════════════════════════════════════════════╣ ║ UPDATE ║ ║ client.update_session(name, metadata) → Modify session ║ ║ client.update_deck(deck_content) → Replace deck ║ ║ client.send_message(refinement) → Refine via chat ║ ╠══════════════════════════════════════════════════════════════════╣ ║ DELETE ║ ║ client.clear_conversation() → Clear history ║ ║ client.delete_session(session_id) → Remove session ║ ╚══════════════════════════════════════════════════════════════════╝ """)

---

📚 Next Steps¶

Now that you're familiar with CRUD operations:

1. Tutorial 01 - Learn ECLIPSE deck fundamentals
2. Tutorial 04 - Explore conversational model building
3. Tutorial 09 - See the complete workflow
4. Tutorial 10 - Deep dive into API details

Resources¶

• 📖 CLARISSA Documentation
• 🐙 GitHub Repository
• 🦊 GitLab Repository
• 📝 ADR-027: Repository Sync

---

CLARISSA - Conversational Language Agent for Reservoir Integrated Simulation System Analysis