Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.praison.ai/llms.txt

Use this file to discover all available pages before exploring further.

Protocols define extension points for custom authentication, pairing stores, and session binding implementations.

​
Quick Start

1

Import Protocols

from praisonaiagents.gateway.protocols import AuthProtocol, PairingProtocol, SessionBindingProtocol
2

Implement Custom Store

import redis
import secrets
from typing import List, Dict, Any, Optional

class RedisPairingStore:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def generate_code(self, channel_type: str = "unknown", channel_id: Optional[str] = None) -> str:
        code = secrets.token_hex(4).upper()  # 8-character hex
        self.redis.setex(f"pairing:{code}", 300, f"{channel_type}:{channel_id or ''}")
        return code
    
    def approve(self, channel_type: str, code: str, user_id: str = "", user_name: str = "") -> bool:
        stored = self.redis.get(f"pairing:{code}")
        if not stored:
            return False
        
        self.redis.delete(f"pairing:{code}")
        self.redis.set(f"paired:{channel_type}:{user_id}", user_name)
        return True
    
    def is_paired(self, channel_id: str, channel_type: str) -> bool:
        return bool(self.redis.exists(f"paired:{channel_type}:{channel_id}"))
    
    def list_paired(self) -> List["PairedChannel"]:
        # Minimal implementation for protocol compliance
        keys = self.redis.keys("paired:*")
        return []  # Return empty for simplicity
    
    def revoke(self, channel_id: str, channel_type: str) -> bool:
        return bool(self.redis.delete(f"paired:{channel_type}:{channel_id}"))
    
    def list_pending(self, channel_type: Optional[str] = None) -> List[Dict[str, Any]]:
        # Minimal implementation for protocol compliance
        return []  # Return empty for simplicity

# Use as protocol-compatible instance
store = RedisPairingStore(redis.Redis())

​
Protocol Specifications

​
AuthProtocol

Authentication implementations for gateway requests.
MethodSignatureReturns
checkcheck(request: Any) -> Dict[str, Any]{ success: bool, user_id: Optional[str], role: Optional[str], metadata: Dict[str, Any] }

​
PairingProtocol

Channel pairing and authorization management.
MethodSignatureReturns
generate_codegenerate_code(channel_type: str = "unknown", channel_id: Optional[str] = None) -> strThe generated pairing code
approveapprove(channel_type: str, code: str, user_id: str = "", user_name: str = "") -> boolTrue if approval succeeded
is_pairedis_paired(channel_id: str, channel_type: str) -> boolWhether the channel is authorised
list_pairedlist_paired() -> List["PairedChannel"]All paired channels
revokerevoke(channel_id: str, channel_type: str) -> boolTrue if revocation succeeded
list_pendinglist_pending(channel_type: Optional[str] = None) -> List[Dict[str, any]]Pending requests with channel, code, user info, age

​
SessionBindingProtocol

Session to principal mapping for state tracking.
MethodSignatureReturns
bindbind(session_id: str, principal: Dict[str, Any]) -> None—
lookuplookup(session_id: str) -> Optional[Dict[str, Any]]Principal info if found
These are @runtime_checkable typing.Protocol classes — any object whose method shapes match (duck typing) satisfies them. No inheritance required.
Importing the Protocol classes is via from praisonaiagents.gateway.protocols import AuthProtocol, PairingProtocol, SessionBindingProtocol.

​
Custom Implementation Example

from praisonaiagents.gateway.protocols import PairingProtocol
import sqlite3
import secrets
import time
from typing import List, Dict, Any, Optional

class SQLitePairingStore:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute('''
            CREATE TABLE IF NOT EXISTS pending_codes (
                code TEXT PRIMARY KEY,
                channel_type TEXT,
                channel_id TEXT,
                created_at INTEGER
            )
        ''')
        conn.execute('''
            CREATE TABLE IF NOT EXISTS paired_channels (
                channel_id TEXT,
                channel_type TEXT,
                user_name TEXT,
                paired_at INTEGER,
                PRIMARY KEY (channel_id, channel_type)
            )
        ''')
        conn.commit()
        conn.close()
    
    def generate_code(self, channel_type: str = "unknown", channel_id: Optional[str] = None) -> str:
        code = secrets.token_hex(4).upper()  # 8-character hex
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            'INSERT OR REPLACE INTO pending_codes VALUES (?, ?, ?, ?)',
            (code, channel_type, channel_id or "", int(time.time()))
        )
        conn.commit()
        conn.close()
        return code
    
    def approve(self, channel_type: str, code: str, user_id: str = "", user_name: str = "") -> bool:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            'SELECT channel_id FROM pending_codes WHERE code = ? AND channel_type = ?',
            (code, channel_type)
        )
        result = cursor.fetchone()
        
        if not result:
            conn.close()
            return False
        
        channel_id = result[0] or user_id
        
        # Remove pending code
        conn.execute('DELETE FROM pending_codes WHERE code = ?', (code,))
        
        # Add to paired channels
        conn.execute(
            'INSERT OR REPLACE INTO paired_channels VALUES (?, ?, ?, ?)',
            (channel_id, channel_type, user_name, int(time.time()))
        )
        
        conn.commit()
        conn.close()
        return True
    
    def is_paired(self, channel_id: str, channel_type: str) -> bool:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            'SELECT 1 FROM paired_channels WHERE channel_id = ? AND channel_type = ?',
            (channel_id, channel_type)
        )
        result = cursor.fetchone() is not None
        conn.close()
        return result
    
    def list_paired(self) -> List["PairedChannel"]:
        # Return empty list for simplicity - implement based on your PairedChannel structure
        return []
    
    def revoke(self, channel_id: str, channel_type: str) -> bool:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            'DELETE FROM paired_channels WHERE channel_id = ? AND channel_type = ?',
            (channel_id, channel_type)
        )
        success = cursor.rowcount > 0
        conn.commit()
        conn.close()
        return success
    
    def list_pending(self, channel_type: Optional[str] = None) -> List[Dict[str, Any]]:
        conn = sqlite3.connect(self.db_path)
        if channel_type:
            cursor = conn.execute(
                'SELECT code, channel_type, channel_id, created_at FROM pending_codes WHERE channel_type = ?',
                (channel_type,)
            )
        else:
            cursor = conn.execute('SELECT code, channel_type, channel_id, created_at FROM pending_codes')
        
        results = []
        for row in cursor.fetchall():
            results.append({
                "code": row[0],
                "channel_type": row[1], 
                "channel_id": row[2],
                "age_seconds": int(time.time()) - row[3]
            })
        conn.close()
        return results

# Use the custom store
custom_store = SQLitePairingStore("pairing.db")

​
How It Works

Protocol implementations allow you to plug custom backends into the PraisonAI gateway without modifying core code.
ProtocolPurposeExtension Point
AuthProtocolRequest authenticationCustom auth backends (LDAP, OAuth, etc.)
PairingProtocolChannel pairing storageCustom storage (Redis, DB, etc.)
SessionBindingProtocolSession managementCustom session stores

​
Configuration Options

Protocol implementations are dependency-injected into the gateway: 
from praisonaiagents.gateway import create_gateway
from your_module import MyCustomPairingStore

# Inject your custom implementation
gateway = create_gateway(
    pairing_store=MyCustomPairingStore(),
    # Other protocols...
)

​
Common Patterns

# Fast, distributed storage for high-traffic scenarios
class RedisPairingStore:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
Best for: Multi-instance deployments, high throughput.

​
Best Practices

Protocol compliance requires implementing all methods, even if some return empty results for your use case.
def list_pending(self, channel_type: Optional[str] = None) -> List[Dict[str, Any]]:
    # Even if you don't need pending logic, implement the method
    return []
Wrap database operations in try-catch blocks and return safe defaults.
def is_paired(self, channel_id: str, channel_type: str) -> bool:
    try:
        # Database operation
        return self._check_database(channel_id, channel_type)
    except Exception as e:
        logger.error(f"Database error: {e}")
        return False  # Fail closed
For database backends, use connection pools to avoid connection overhead.
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    db_url,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)
Always validate and sanitize inputs, especially for database queries.
def approve(self, channel_type: str, code: str, user_id: str = "", user_name: str = "") -> bool:
    # Validate inputs
    if not code.isalnum() or len(code) != 8:
        return False
    # Use parameterized queries
    return self._execute_query("UPDATE ...", (channel_type, code))

Bot Pairing

Complete pairing system documentation

Unknown User Pairing

Owner-DM inline-button approval flow