yumoqing/pipeline_core

Fork 0

yumoqing 1abaa4ea2e feat: 技能库(skills_library 213个Hermes技能)+skill_pack技能集安装能力+ocai-h5-dev技能集(37个应用/模块开发技能)

2026-08-16 15:17:43 +08:00

28 KiB

Raw Blame History

name

version

description

trigger_conditions

hermes-service-module-implementation

1.0.0

Complete production-ready implementation of Hermes Service web application that provides API access to Hermes Agent CLI functionality while maintaining upgrade compatibility and security.

User requests to create a web service that exposes Hermes Agent CLI functionality via APIs

Task involves creating a production-ready service that can be safely upgraded

Need to provide API access to Hermes tools while maintaining security isolation

Hermes Service Module Implementation Guide

This skill provides a complete implementation pattern for creating Hermes Service modules that extend Hermes Agent functionality through web APIs while maintaining proper multi-user isolation and following established development conventions.

Key Principles

Directory Structure

Place service modules in ~/repos/hermes-service/
Use clean user data structure: /d/hermesai/users/{user_id}/.hermes
Follow standard Python package layout with FastAPI backend

Multi-User Isolation Strategy

Dynamic User Creation: Automatically create isolated environments for new users
Data Separation: Each user gets independent .hermes/ directory with separate state.db
Resource Sharing: Share virtual environment to save disk space while maintaining isolation
Environment Variables: Use HOME environment variable to redirect Hermes to user-specific directories

API Design Patterns

Session Management: Create sessions with user context and message history
Command Execution: Execute Hermes CLI commands in isolated user contexts
Error Handling: Proper HTTP status codes and error propagation
Security: Bind to localhost only, implement proper authentication layer This skill provides a complete implementation guide for creating a Hermes Service web application that exposes Hermes Agent CLI functionality through standardized REST APIs. The service runs independently from the main Hermes Dashboard but leverages the existing Hermes Agent installation and virtual environment.

Architecture Principles

Core Design Decisions

Independent Web Service: Separate FastAPI application that calls existing Hermes CLI commands
True Multi-User Isolation: Each user gets completely isolated Hermes environment with independent state.db and configuration
Persistent User Data: User data stored in /d/hermesai/.hermes/users/ (within Hermes installation directory) with 700 permissions
Resource Efficient: Shared virtual environment via symbolic links to avoid disk space duplication
Upgrade Safe: Service can be upgraded independently without affecting Hermes Agent core functionality
Security First: Binds to localhost by default, includes timeout protection, and integrates with rbac for authentication
Production Ready: Includes health checks, proper error handling, and comprehensive API documentation

Multi-User Isolation Strategy

The key innovation is using the HOME environment variable to redirect Hermes Agent to user-specific directories:

User Environment Path: /d/hermesai/.hermes/users/user-{user_id}/
Isolated .hermes Directory: Each user gets their own .hermes/ folder containing state.db, sessions, and config
Environment Variable: env['HOME'] = user_hermes_dot_path ensures complete data isolation
Security: 700 permissions on all user directories prevent cross-user access
Sanitization: User IDs are sanitized to prevent directory traversal attacks

Integration Strategy

Leverages Existing Installation: Uses Hermes Agent's virtual environment and Python path
CLI Command Execution: Executes actual hermes CLI commands rather than reimplementing logic
Optional rbac Integration: Can integrate with existing rbac module for user authentication if needed
Extensible Design: Supports adding database persistence, WebSocket support, and advanced features

Implementation Steps

1. Create Service Directory Structure

mkdir -p ~/repos/hermes-service/

2. Implement Main Service File

Create ~/repos/hermes-service/main.py with FastAPI application that includes true multi-user isolation:

#!/usr/bin/env python3
"""
Hermes Service with true multi-user support using persistent user directories
"""

import os
import sys
import asyncio
import uuid
from datetime import datetime
from pathlib import Path
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
import json
import shutil

# Base Hermes Agent path
BASE_HERMES_PATH = "/d/hermesai/.hermes/hermes-agent"

# User data directory within Hermes installation
# This follows the pattern: ~/.hermes/users/
HERMES_DOT_PATH = "/d/hermesai/.hermes"
USER_HERMES_BASE = os.path.join(HERMES_DOT_PATH, "users")

def get_user_hermes_path(user_id: str) -> str:
    """Get isolated Hermes environment path for a user"""
    if not user_id or user_id == "anonymous":
        user_id = "anonymous"
    # Sanitize user_id to prevent directory traversal
    safe_user_id = "".join(c for c in user_id if c.isalnum() or c in "-_.")
    return os.path.join(USER_HERMES_BASE, f"user-{safe_user_id}")

def ensure_user_hermes_env(user_id: str):
    """Ensure user has isolated Hermes environment"""
    user_hermes_path = get_user_hermes_path(user_id)
    
    if not os.path.exists(user_hermes_path):
        # Create user directory with proper permissions
        os.makedirs(user_hermes_path, exist_ok=True, mode=0o700)
        
        # Copy base Hermes files (excluding .git, __pycache__, etc.)
        shutil.copytree(
            BASE_HERMES_PATH,
            user_hermes_path,
            dirs_exist_ok=True,
            ignore=shutil.ignore_patterns('.git', '__pycache__', '*.pyc', '.venv', 'web_dist')
        )
        
        # Create isolated .hermes directory
        user_dot_hermes = os.path.join(user_hermes_path, '.hermes')
        os.makedirs(user_dot_hermes, exist_ok=True, mode=0o700)
        
        # Create symbolic link to shared virtual environment
        venv_link = os.path.join(user_hermes_path, '.venv')
        if not os.path.exists(venv_link):
            os.symlink(
                os.path.join(BASE_HERMES_PATH, '.venv'),
                venv_link
            )
    
    return user_hermes_path

async def execute_hermes_command(command_args, user_id=None, timeout=300):
    """Execute hermes CLI command in isolated user environment"""
    try:
        # Get user-specific Hermes environment
        if user_id:
            user_hermes_path = ensure_user_hermes_env(user_id)
            hermes_dot_path = os.path.join(user_hermes_path, '.hermes')
        else:
            user_hermes_path = BASE_HERMES_PATH
            hermes_dot_path = HERMES_DOT_PATH
        
        python_path = "/d/hermesai/.hermes/hermes-agent/.venv/bin/python3"
        cmd = [python_path, "-m", "hermes_cli.main"] + command_args
        
        # Set environment for isolated execution - THIS IS THE KEY
        env = os.environ.copy()
        env['HOME'] = hermes_dot_path  # This makes Hermes use isolated .hermes directory
        env['HERMES_USER_ID'] = str(user_id or 'anonymous')
        env['HERMES_SESSION_ID'] = str(uuid.uuid4())
        
        process = await asyncio.create_subprocess_exec(
            *cmd,
            cwd=user_hermes_path,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        
        try:
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
            return {
                'success': process.returncode == 0,
                'stdout': stdout.decode('utf-8', errors='replace'),
                'stderr': stderr.decode('utf-8', errors='replace'),
                'returncode': process.returncode
            }
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            return {
                'success': False,
                'stdout': '',
                'stderr': f'Command timed out after {timeout} seconds',
                'returncode': -1
            }
            
    except Exception as e:
        return {
            'success': False,
            'stdout': '',
            'stderr': str(e),
            'returncode': -1
        }

# Create the base directory structure
os.makedirs(USER_HERMES_BASE, exist_ok=True, mode=0o700)

# ... rest of FastAPI app with session management and WebSocket endpoints

3. Create User Directory Structure

mkdir -p /d/hermesai/.hermes/users
chmod 700 /d/hermesai/.hermes/users

"""

import os import sys import asyncio import uuid from datetime import datetime from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional, Dict, Any, List import json

Add Hermes Agent to Python path

HERMES_PATH = "/d/hermesai/.hermes/hermes-agent" sys.path.insert(0, HERMES_PATH)

app = FastAPI(title="Hermes Service API", version="1.0.1")

Configure CORS

app.add_middleware( CORSMiddleware, allow_origins=[""], allow_credentials=True, allow_methods=[""], allow_headers=["*"], )

In-memory session storage (in production, use Redis or database)

active_sessions = {}

class CommandRequest(BaseModel): command: list[str] user_context: Optional[Dict[str, Any]] = None timeout: int = 300

class SessionCreateRequest(BaseModel): user_id: Optional[str] = None initial_message: Optional[str] = None

class SessionMessage(BaseModel): session_id: str message: str user_context: Optional[Dict[str, Any]] = None

@app.get("/health") async def health_check(): """Health check endpoint""" return {"status": "healthy", "service": "hermes-service"}

@app.get("/api/v1/status") async def get_hermes_status(): """Get Hermes Agent status""" try: result = await execute_hermes_command(["--version"]) return {"status": "running", "version": result.get("stdout", "").strip()} except Exception as e: return {"status": "error", "error": str(e)}

@app.post("/api/v1/sessions") async def create_session(request: SessionCreateRequest): """Create a new interactive session""" session_id = str(uuid.uuid4()) session_data = { "id": session_id, "user_id": request.user_id, "created_at": datetime.now().isoformat(), "messages": [], "status": "active" }

if request.initial_message:
    session_data["messages"].append({
        "role": "user",
        "content": request.initial_message,
        "timestamp": datetime.now().isoformat()
    })

active_sessions[session_id] = session_data
return {"session_id": session_id, "status": "created"}

@app.post("/api/v1/sessions/{session_id}/messages") async def send_message(session_id: str, request: SessionMessage): """Send a message to an existing session (non-streaming)""" if session_id not in active_sessions: raise HTTPException(status_code=404, detail="Session not found")

# Add user message to session
active_sessions[session_id]["messages"].append({
    "role": "user",
    "content": request.message,
    "timestamp": datetime.now().isoformat()
})

# Execute the message as a hermes command
command_args = ["chat", request.message]
result = await execute_hermes_command(
    command_args,
    user_context=request.user_context
)

# Add assistant response to session
response_content = result.get("stdout", "") if result["success"] else result.get("stderr", "Command failed")
active_sessions[session_id]["messages"].append({
    "role": "assistant",
    "content": response_content,
    "timestamp": datetime.now().isoformat()
})

return {
    "session_id": session_id,
    "response": response_content,
    "success": result["success"]
}

@app.websocket("/api/v1/sessions/{session_id}/stream") async def stream_session(websocket: WebSocket, session_id: str): """WebSocket endpoint for real-time streaming interaction""" await websocket.accept()

try:
    # Create session if it doesn't exist
    if session_id not in active_sessions:
        active_sessions[session_id] = {
            "id": session_id,
            "created_at": datetime.now().isoformat(),
            "messages": [],
            "status": "active"
        }
    
    while True:
        # Receive message from client
        data = await websocket.receive_text()
        try:
            message_data = json.loads(data)
            user_message = message_data.get("message", "")
            user_context = message_data.get("user_context", {})
            
            if not user_message:
                await websocket.send_text(json.dumps({"error": "Empty message"}))
                continue
            
            # Add user message to session
            active_sessions[session_id]["messages"].append({
                "role": "user",
                "content": user_message,
                "timestamp": datetime.now().isoformat()
            })
            
            # Stream the hermes command execution
            await stream_hermes_command(
                websocket,
                ["chat", user_message],
                user_context=user_context,
                session_id=session_id
            )
            
        except json.JSONDecodeError:
            await websocket.send_text(json.dumps({"error": "Invalid JSON format"}))
        except Exception as e:
            await websocket.send_text(json.dumps({"error": str(e)}))
            
except WebSocketDisconnect:
    print(f"WebSocket disconnected for session {session_id}")
except Exception as e:
    await websocket.send_text(json.dumps({"error": f"Connection error: {str(e)}"}))

async def execute_hermes_command(command_args, user_context=None, timeout=300): """Execute hermes CLI command with user context""" try: python_path = "/d/hermesai/.hermes/hermes-agent/.venv/bin/python3" cmd = [python_path, "-m", "hermes_cli.main"] + command_args

    env = os.environ.copy()
    if user_context:
        env['HERMES_USER_ID'] = str(user_context.get('user_id', ''))
        env['HERMES_SESSION_ID'] = str(user_context.get('session_id', ''))
    
    process = await asyncio.create_subprocess_exec(
        *cmd,
        cwd=HERMES_PATH,
        env=env,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    
    try:
        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
        return {
            'success': process.returncode == 0,
            'stdout': stdout.decode('utf-8', errors='replace'),
            'stderr': stderr.decode('utf-8', errors='replace'),
            'returncode': process.returncode
        }
    except asyncio.TimeoutError:
        process.kill()
        await process.wait()
        return {
            'success': False,
            'stdout': '',
            'stderr': f'Command timed out after {timeout} seconds',
            'returncode': -1
        }
        
except Exception as e:
    return {
        'success': False,
        'stdout': '',
        'stderr': str(e),
        'returncode': -1
    }

async def stream_hermes_command(websocket, command_args, user_context=None, session_id=None, timeout=300): """Stream hermes command execution in real-time""" try: python_path = "/d/hermesai/.hermes/hermes-agent/.venv/bin/python3" cmd = [python_path, "-m", "hermes_cli.main"] + command_args

    env = os.environ.copy()
    if user_context:
        env['HERMES_USER_ID'] = str(user_context.get('user_id', ''))
        env['HERMES_SESSION_ID'] = str(user_context.get('session_id', ''))
    
    process = await asyncio.create_subprocess_exec(
        *cmd,
        cwd=HERMES_PATH,
        env=env,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    
    # Stream stdout in real-time
    async def read_stream(stream, stream_name):
        while True:
            line = await stream.readline()
            if not line:
                break
            line_str = line.decode('utf-8', errors='replace')
            await websocket.send_text(json.dumps({
                "type": "output",
                "stream": stream_name,
                "data": line_str,
                "session_id": session_id
            }))
    
    # Start streaming both stdout and stderr
    stdout_task = asyncio.create_task(read_stream(process.stdout, "stdout"))
    stderr_task = asyncio.create_task(read_stream(process.stderr, "stderr"))
    
    try:
        await asyncio.wait_for(asyncio.gather(stdout_task, stderr_task), timeout=timeout)
        returncode = await process.wait()
        
        await websocket.send_text(json.dumps({
            "type": "complete",
            "returncode": returncode,
            "session_id": session_id
        }))
        
    except asyncio.TimeoutError:
        process.kill()
        await process.wait()
        await websocket.send_text(json.dumps({
            "type": "error",
            "message": f"Command timed out after {timeout} seconds",
            "session_id": session_id
        }))
        
except Exception as e:
    await websocket.send_text(json.dumps({
        "type": "error",
        "message": str(e),
        "session_id": session_id
    }))

if name == "main": import uvicorn uvicorn.run(app, host="127.0.0.1", port=9120, log_level="info")

Hermes Service - Web API wrapper for Hermes Agent CLI functionality
"""

import os
import sys
import subprocess
import asyncio
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, Dict, Any

# Add Hermes Agent to Python path
HERMES_PATH = "/d/hermesai/.hermes/hermes-agent"
sys.path.insert(0, HERMES_PATH)

app = FastAPI(title="Hermes Service API", version="1.0.0")

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class CommandRequest(BaseModel):
    command: list[str]
    user_context: Optional[Dict[str, Any]] = None
    timeout: int = 300

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "hermes-service"}

@app.get("/api/v1/status")
async def get_hermes_status():
    """Get Hermes Agent status"""
    try:
        result = await execute_hermes_command(["--version"])
        return {"status": "running", "version": result.get("stdout", "").strip()}
    except Exception as e:
        return {"status": "error", "error": str(e)}

@app.post("/api/v1/execute")
async def execute_command(request: CommandRequest):
    """Execute Hermes CLI command"""
    try:
        result = await execute_hermes_command(
            request.command,
            user_context=request.user_context,
            timeout=request.timeout
        )
        if not result["success"]:
            raise HTTPException(status_code=500, detail=result["stderr"])
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

async def execute_hermes_command(command_args, user_context=None, timeout=300):
    """Execute hermes CLI command with user context"""
    try:
        # Use the virtual environment Python
        python_path = "/d/hermesai/.hermes/hermes-agent/.venv/bin/python3"
        cmd = [python_path, "-m", "hermes_cli.main"] + command_args
        
        env = os.environ.copy()
        if user_context:
            env['HERMES_USER_ID'] = str(user_context.get('user_id', ''))
            env['HERMES_SESSION_ID'] = str(user_context.get('session_id', ''))
        
        # Run command with timeout
        process = await asyncio.create_subprocess_exec(
            *cmd,
            cwd=HERMES_PATH,
            env=env,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        
        try:
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
            return {
                'success': process.returncode == 0,
                'stdout': stdout.decode('utf-8', errors='replace'),
                'stderr': stderr.decode('utf-8', errors='replace'),
                'returncode': process.returncode
            }
        except asyncio.TimeoutError:
            process.kill()
            await process.wait()
            return {
                'success': False,
                'stdout': '',
                'stderr': f'Command timed out after {timeout} seconds',
                'returncode': -1
            }
            
    except Exception as e:
        return {
            'success': False,
            'stdout': '',
            'stderr': str(e),
            'returncode': -1
        }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=9120, log_level="info")

3. Install Dependencies

Use the existing Hermes Agent virtual environment:

cd ~/repos/hermes-service
/d/hermesai/.hermes/hermes-agent/.venv/bin/python3 -m pip install fastapi uvicorn[standard] python-dotenv

4. Start the Service

cd ~/repos/hermes-service
/d/hermesai/.hermes/hermes-agent/.venv/bin/python3 main.py

API Documentation

Web Service Configuration

Default Port: 9120
Protocol: HTTP/WebSocket
Base URL: http://localhost:9120
Host Binding: 127.0.0.1 (localhost only for security)

API Endpoints

Health Check

Endpoint: GET /health
Response: {"status": "healthy", "service": "hermes-service"}
Auth Required: No

Hermes Status

Endpoint: GET /api/v1/status
Response: {"status": "running", "version": "Hermes Agent v0.10.0"}
Auth Required: No

Command Execution

Endpoint: POST /api/v1/execute

Request Body:

{
  "command": ["terminal", "ls", "-la"],
  "user_context": {"user_id": "123", "session_id": "456"},
  "timeout": 300
}

Response:

{
  "success": true,
  "stdout": "command output",
  "stderr": "",
  "returncode": 0
}

Session Management

Create Session: POST /api/v1/sessions
- Request: {"user_id": "user123", "initial_message": "hello"}
- Response: {"session_id": "uuid", "status": "created"}
Send Message: POST /api/v1/sessions/{session_id}/messages
- Request: {"message": "what is your name?", "user_context": {"user_id": "user123"}}
- Response: {"session_id": "uuid", "response": "I am Hermes...", "success": true}

Real-time Streaming

WebSocket Endpoint: ws://localhost:9120/api/v1/sessions/{session_id}/stream

Message Format (client → server):

{
  "message": "terminal ls -la",
  "user_context": {"user_id": "user123"}
}

Stream Format (server → client):

{
  "type": "output",
  "stream": "stdout",
  "data": "file listing...\n",
  "session_id": "uuid"
}

Completion Format:

{
  "type": "complete",
  "returncode": 0,
  "session_id": "uuid"
}

Security Considerations

Built-in Protections

Local Binding: Service only accessible from localhost by default
Timeout Protection: Commands automatically timeout after 5 minutes
Error Isolation: Failed commands don't crash the service
Input Validation: Command arguments are passed directly to CLI (no shell injection)

Production Enhancements

Authentication: Integrate rbac module for JWT token validation
Rate Limiting: Add request rate limiting per user
Command Whitelisting: Restrict which CLI commands can be executed
Audit Logging: Log all command executions with user context

Upgrade Compatibility

Zero-Downtime Upgrades

The service architecture supports safe upgrades because:

Independent Process: Service runs separately from Hermes Agent
API Versioning: Use /api/v1/ prefix for backward compatibility
Graceful Degradation: Failed commands return proper error responses
Rolling Updates: Multiple service instances can run during upgrades

Deployment with Systemd

Create systemd service file for automatic startup:

[Unit]
Description=Hermes Service API
After=network.target

[Service]
Type=simple
WorkingDirectory=/d/hermesai/repos/hermes-service
ExecStart=/d/hermesai/.hermes/hermes-agent/.venv/bin/python3 main.py
Restart=always
RestartSec=10
User=hermesai

[Install]
WantedBy=default.target

Extension Points

Database Integration

Add SQL database support for:

Session persistence: Replace in-memory sessions with database storage following database-table-definition-spec
Service instance management: Store multiple Hermes service configurations per user
User preferences and configurations
Command execution history and audit logs
Multi-user session isolation with proper rbac integration

Enhanced WebSocket Support

The current implementation already includes real-time streaming, but can be extended with:

Bidirectional tool approval workflows: Handle dangerous command approvals through WebSocket
Session state synchronization: Sync session state across multiple clients
File transfer capabilities: Stream file uploads/downloads for tool operations
Rich media support: Handle images, audio, and other multimedia responses

rbac Integration

Integrate with existing rbac module for comprehensive security:

from rbac.check_perm import check_permission

def verify_user_access(user_id: str, permission: str) -> bool:
    return check_permission(user_id, 'hermes_service', permission)

# Apply to all endpoints requiring authentication
@app.post("/api/v1/sessions")
async def create_session(request: SessionCreateRequest):
    if request.user_id and not verify_user_access(request.user_id, 'create_session'):
        raise HTTPException(status_code=403, detail="Insufficient permissions")
    # ... rest of implementation

Production Session Management

Replace in-memory sessions with production-ready storage:

Redis: For fast session access and TTL-based cleanup
PostgreSQL: For persistent session storage with full CRUD operations
Session cleanup: Implement background tasks to remove expired sessions
Maximum session limits: Prevent resource exhaustion per user

Verification Steps

Service starts successfully on port 9120
Health check endpoint returns valid response
Hermes status endpoint shows correct version
Command execution works with various Hermes CLI commands
Timeout protection works for long-running commands
Error handling properly catches and reports failures
Service can be managed with systemd (optional)
API documentation matches actual behavior

Common Pitfalls and Solutions

Pitfall 1: Python Path Issues

Problem: Service can't find Hermes Agent modules Solution: Explicitly add Hermes Agent path to sys.path and use virtual environment Python

Pitfall 2: Command Hanging

Problem: Long-running commands block the service Solution: Implement asyncio timeout with proper process cleanup

Pitfall 3: Encoding Issues

Problem: Non-UTF8 output causes crashes Solution: Use errors='replace' in decode() calls

Pitfall 4: Security Exposure

Problem: Service accidentally exposed to network Solution: Bind to 127.0.0.1 by default, require explicit configuration for external access

28 KiB Raw Blame History

Hermes Service Module Implementation Guide

Key Principles

Directory Structure

Multi-User Isolation Strategy

API Design Patterns

Architecture Principles

Core Design Decisions

Multi-User Isolation Strategy

Integration Strategy

Implementation Steps

1. Create Service Directory Structure

2. Implement Main Service File

3. Create User Directory Structure

Add Hermes Agent to Python path

Configure CORS

In-memory session storage (in production, use Redis or database)

3. Install Dependencies

4. Start the Service

API Documentation

Web Service Configuration

API Endpoints

Health Check

Hermes Status

Command Execution

Session Management

Real-time Streaming

Security Considerations

Built-in Protections

Production Enhancements

Upgrade Compatibility

Zero-Downtime Upgrades

Deployment with Systemd

Extension Points

Database Integration

Enhanced WebSocket Support

rbac Integration

Production Session Management

Verification Steps

Common Pitfalls and Solutions

Pitfall 1: Python Path Issues

Pitfall 2: Command Hanging

Pitfall 3: Encoding Issues

Pitfall 4: Security Exposure

28 KiB

Raw Blame History