Custom Auth & Pairing Protocols

Plug in your own auth, pairing store, or session binding using runtime-checkable Protocols

Plug custom auth, pairing stores, or session binding into your gateway via runtime-checkable Protocols.

from praisonaiagents import Agent

agent = Agent(name="pairing-agent", instructions="Use pairing protocols for agent handoff.")
agent.start("Pair this conversation session with a specialist agent.")

The user pairs through the gateway; custom Protocol implementations back auth, pairing codes, and session binding.

from praisonaiagents.gateway.protocols import PairingProtocol

class MyPairingStore:
    def generate_code(self, channel_type: str = "unknown", channel_id=None) -> str: ...
    def approve(self, channel_type: str, code: str, user_id: str = "", user_name: str = "") -> bool: ...
    def is_paired(self, channel_id: str, channel_type: str) -> bool: ...
    def list_paired(self): ...
    def revoke(self, channel_id: str, channel_type: str) -> bool: ...
    def list_pending(self, channel_type=None): ...

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.

Method	Signature	Returns
check	check(request: Any) -> Dict[str, Any]	{ success: bool, user_id: Optional[str], role: Optional[str], metadata: Dict[str, Any] }

PairingProtocol

Channel pairing and authorization management.

Method	Signature	Returns
generate_code	generate_code(channel_type: str = "unknown", channel_id: Optional[str] = None) -> str	The generated pairing code
approve	approve(channel_type: str, code: str, user_id: str = "", user_name: str = "") -> bool	True if approval succeeded
is_paired	is_paired(channel_id: str, channel_type: str) -> bool	Whether the channel is authorised
list_paired	list_paired() -> List["PairedChannel"]	All paired channels
revoke	revoke(channel_id: str, channel_type: str) -> bool	True if revocation succeeded
list_pending	list_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.

Method	Signature	Returns
bind	bind(session_id: str, principal: Dict[str, Any]) -> None	—
lookup	lookup(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.

Protocol	Purpose	Extension Point
AuthProtocol	Request authentication	Custom auth backends (LDAP, OAuth, etc.)
PairingProtocol	Channel pairing storage	Custom storage (Redis, DB, etc.)
SessionBindingProtocol	Session management	Custom session stores

---

Configuration Options

Protocol implementations are injected when you start the gateway:

from praisonai.gateway import WebSocketGateway
from your_module import MyCustomPairingStore

gateway = WebSocketGateway(port=8765)
# Pass your store when wiring the bot adapter (see Bot Pairing)
store = MyCustomPairingStore()

---

Common Patterns

Redis Backend

# 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.

Database Backend

# Persistent storage with ACID guarantees
class DBPairingStore:
    def __init__(self, db_url: str):
        self.engine = create_engine(db_url)

Best for: Audit trails, complex queries, data persistence.

File Backend

# Simple file-based storage for development
class FilePairingStore:
    def __init__(self, file_path: str):
        self.file_path = file_path

Best for: Development, single-instance deployments.

---

Best Practices

Implement all protocol methods

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 []

Handle database errors gracefully

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

Use connection pooling for performance

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
)

Validate inputs to prevent injection

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))

---

Related

Bot Pairing

Complete pairing system documentation

Unknown User Pairing

Owner-DM inline-button approval flow