# Stateful Sessions

> Practical patterns for using sessions in real applications.

**URL:** https://inference.sh/docs/examples/stateful-sessions
**Last updated:** 2026-09-18

---

Practical patterns for using sessions in real applications.

---

## Browser Automation

Control a browser across multiple steps:

```python
from inferencesh import Inference

client = Inference(api_key="...")

# Start browser session with 10-minute timeout (browser interactions can be slow)
result = client.run({
    "app": "browser-automation@abc123",
    "function": "navigate",
    "input": {"url": "https://example.com"},
    "session": "new",
    "session_timeout": 600  # 10 minutes
})
session_id = result["session_id"]

try:
    # Click login button
    client.run({
        "app": "browser-automation@abc123",
        "function": "click",
        "input": {"selector": "#login-btn"},
        "session": session_id
    })

    # Fill form
    client.run({
        "app": "browser-automation@abc123",
        "function": "type",
        "input": {"selector": "#email", "text": "user@example.com"},
        "session": session_id
    })

    # Take screenshot
    result = client.run({
        "app": "browser-automation@abc123",
        "function": "screenshot",
        "input": {},
        "session": session_id
    })
    print(f"Screenshot: {result['output']['uri']}")

finally:
    client.sessions.end(session_id)
```

---

## Code Interpreter

Run code with accumulated state:

```python
from inferencesh import Inference

client = Inference(api_key="...")

# Start interpreter session
result = client.run({
    "app": "python-interpreter@abc123",
    "function": "run",
    "input": {"code": "import pandas as pd\ndf = pd.DataFrame({'a': [1,2,3]})"},
    "session": "new"
})
session_id = result["session_id"]

try:
    # Variables persist between calls
    result = client.run({
        "app": "python-interpreter@abc123",
        "function": "run",
        "input": {"code": "df['b'] = df['a'] * 2"},
        "session": session_id
    })

    # Access previously defined data
    result = client.run({
        "app": "python-interpreter@abc123",
        "function": "run",
        "input": {"code": "print(df.to_string())"},
        "session": session_id
    })
    print(result["output"]["stdout"])

finally:
    client.sessions.end(session_id)
```

---

## Interactive ML Pipeline

Load model once, run multiple inferences:

```python
from inferencesh import Inference

client = Inference(api_key="...")

# Initialize model with 30-minute timeout (model loading is expensive)
result = client.run({
    "app": "ml-pipeline@abc123",
    "function": "load_model",
    "input": {"model_name": "gpt2-large"},
    "session": "new",
    "session_timeout": 1800  # 30 minutes - keep model loaded longer
})
session_id = result["session_id"]

try:
    # Fast inference calls
    prompts = ["Hello,", "The weather is", "I think that"]

    for prompt in prompts:
        result = client.run({
            "app": "ml-pipeline@abc123",
            "function": "generate",
            "input": {"prompt": prompt, "max_tokens": 50},
            "session": session_id
        })
        print(f"{prompt} → {result['output']['text']}")

finally:
    client.sessions.end(session_id)
```

---

## Distributed Usage

Pass session IDs between services:

### Service A: Start Session

```python
# service_a.py
import redis
from inferencesh import Inference

client = Inference(api_key="...")
r = redis.Redis()

# Start session and store ID
result = client.run({
    "app": "data-processor@abc123",
    "function": "init",
    "input": {"config": "..."},
    "session": "new"
})

# Store session ID for other services
r.set("current_session", result["session_id"])
print(f"Session started: {result['session_id']}")
```

### Service B: Use Session

```python
# service_b.py
import redis
from inferencesh import Inference

client = Inference(api_key="...")
r = redis.Redis()

# Retrieve session ID
session_id = r.get("current_session").decode()

# Continue the session
result = client.run({
    "app": "data-processor@abc123",
    "function": "process",
    "input": {"data": "..."},
    "session": session_id
})
```

### Service C: End Session

```python
# service_c.py
import redis
from inferencesh import Inference

client = Inference(api_key="...")
r = redis.Redis()

session_id = r.get("current_session").decode()

# End session and cleanup
client.sessions.end(session_id)
r.delete("current_session")
```

---

## Context Manager Pattern (Python)

Automatic cleanup with context managers:

```python
from inferencesh import Inference

client = Inference(api_key="...")

# Session automatically ends when exiting the block
with client.session("browser-automation@abc123") as session:
    session.call("navigate", {"url": "https://example.com"})
    session.call("click", {"selector": "#button"})
    result = session.call("screenshot", {})
    print(f"Screenshot: {result['output']['uri']}")
# Session ended automatically

# Also works with exceptions
with client.session("browser-automation@abc123") as session:
    session.call("navigate", {"url": "https://example.com"})
    raise ValueError("Something went wrong")
# Session still ended automatically
```

---

## Error Recovery

Handle session expiration gracefully:

```python
from inferencesh import (
    Inference,
    SessionExpiredError,
    SessionNotFoundError,
)

client = Inference(api_key="...")

class ResilientSession:
    def __init__(self, client, app):
        self.client = client
        self.app = app
        self.session_id = None

    def call(self, function, input):
        if not self.session_id:
            return self._call_new(function, input)

        try:
            return self._call_existing(function, input)
        except (SessionExpiredError, SessionNotFoundError):
            # Session lost — recreate
            return self._call_new(function, input)

    def _call_new(self, function, input):
        result = self.client.run({
            "app": self.app,
            "function": function,
            "input": input,
            "session": "new"
        })
        self.session_id = result["session_id"]
        return result

    def _call_existing(self, function, input):
        return self.client.run({
            "app": self.app,
            "function": function,
            "input": input,
            "session": self.session_id
        })

    def end(self):
        if self.session_id:
            try:
                self.client.sessions.end(self.session_id)
            except Exception:
                pass  # Best effort
            self.session_id = None


# Usage
session = ResilientSession(client, "browser-automation@abc123")
try:
    session.call("navigate", {"url": "https://example.com"})
    # ... long running work ...
    session.call("screenshot", {})  # Auto-recreates if expired
finally:
    session.end()
```

---

## Custom Timeout Examples

Choose the right timeout for your use case:

### Quick Interactions (Short Timeout)

```python
# Short-lived session for quick form filling (2 minutes)
result = client.run({
    "app": "form-filler@abc123",
    "input": {...},
    "session": "new",
    "session_timeout": 120
})
```

### Interactive Workflows (Medium Timeout)

```python
# Interactive session with human-in-the-loop (5 minutes)
result = client.run({
    "app": "assistant@abc123",
    "input": {...},
    "session": "new",
    "session_timeout": 300
})
```

### Long-Running Pipelines (Long Timeout)

```python
# Data processing pipeline (1 hour max)
result = client.run({
    "app": "data-processor@abc123",
    "input": {...},
    "session": "new",
    "session_timeout": 3600  # Maximum allowed
})
```

### Timeout Guidelines

| Use Case | Recommended Timeout |
|----------|---------------------|
| Quick API calls | 30-60 seconds |
| Form filling, simple automation | 2-5 minutes |
| Interactive assistants | 5-15 minutes |
| Browser automation | 10-30 minutes |
| ML model inference | 15-60 minutes |
| Batch processing | 30-60 minutes |

---

## Cleanup Best Practices

### Always Use Try/Finally

```python
result = client.run({..., "session": "new"})
session_id = result["session_id"]

try:
    # Your session work
    pass
finally:
    client.sessions.end(session_id)
```

### Handle Cleanup Errors

```python
def safe_end_session(client, session_id):
    """End session, ignoring errors."""
    try:
        client.sessions.end(session_id)
    except Exception as e:
        print(f"Warning: Failed to end session {session_id}: {e}")
```

### Periodic Cleanup

List and clean up old sessions:

```python
from datetime import datetime, timedelta

def cleanup_old_sessions(client, max_age_hours=1):
    """End sessions older than max_age."""
    cutoff = datetime.utcnow() - timedelta(hours=max_age_hours)

    sessions = client.sessions.list()
    for session in sessions:
        if session.status == "active" and session.created_at < cutoff:
            try:
                client.sessions.end(session.id)
                print(f"Ended old session: {session.id}")
            except Exception as e:
                print(f"Failed to end {session.id}: {e}")
```

---

## Next

- [Sessions Concept](/docs/concepts/sessions) — Overview
- [Sessions Developer Guide](/docs/extend/sessions) — Full API reference