Protocols

Lightweight interfaces for mocking, testing, and custom implementations

Why Protocols?

Protocols enable testing without real LLM calls and custom implementations of agents, memory, and tools.

Use Case	Protocol
Mock agents in tests	AgentProtocol
Custom memory backends	MemoryProtocol
Mock tools for testing	ToolProtocol

---

Quick Start

1. Mock an Agent for Testing

from praisonaiagents.agent.protocols import AgentProtocol

class MockAgent:
    """Use in tests - no LLM costs, instant responses."""
    
    @property
    def name(self) -> str:
        return "TestAgent"
    
    def chat(self, prompt: str, **kwargs) -> str:
        return f"Mock response to: {prompt}"
    
    async def achat(self, prompt: str, **kwargs) -> str:
        return f"Async mock: {prompt}"

# Use in your tests
def test_my_workflow():
    agent = MockAgent()
    result = agent.chat("Hello")
    assert "Mock response" in result

2. Custom Memory Backend

from praisonaiagents.memory.protocols import MemoryProtocol

class RedisMemory:
    """Production memory using Redis."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def store_short_term(self, text: str, metadata=None, **kwargs) -> str:
        key = f"stm:{hash(text)}"
        self.redis.set(key, text, ex=3600)  # 1h TTL
        return key
    
    def search_short_term(self, query: str, limit: int = 5, **kwargs):
        # Implement your search logic
        return []
    
    def store_long_term(self, text: str, metadata=None, **kwargs) -> str:
        key = f"ltm:{hash(text)}"
        self.redis.set(key, text)
        return key
    
    def search_long_term(self, query: str, limit: int = 5, **kwargs):
        return []

# Use with agents
memory = RedisMemory(redis_client)

3. Mock Tools for Testing

from praisonaiagents.tools.protocols import ToolProtocol

class MockSearchTool:
    """Test tool - returns predictable results."""
    
    name = "search"
    description = "Mock search"
    
    def run(self, query: str = "", **kwargs) -> str:
        return f"Found 3 results for: {query}"
    
    def get_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}
            }
        }

# Use in tests
tool = MockSearchTool()
agent = Agent(tools=[tool])

---

All Protocols

AgentProtocol

from praisonaiagents.agent.protocols import (
    AgentProtocol,           # Minimal: name, chat, achat
    RunnableAgentProtocol,   # + run, start, arun, astart
    ToolAwareAgentProtocol,  # + tools property
    MemoryAwareAgentProtocol, # + chat_history, clear_history
    FullAgentProtocol,       # All combined
)

MemoryProtocol

from praisonaiagents.memory.protocols import (
    MemoryProtocol,          # store/search short/long term
    AsyncMemoryProtocol,     # Async variants
    ResettableMemoryProtocol, # + reset methods
    EntityMemoryProtocol,    # + entity storage
)

ToolProtocol

from praisonaiagents.tools.protocols import (
    ToolProtocol,            # name, description, run, get_schema
    CallableToolProtocol,    # + __call__
    AsyncToolProtocol,       # + arun
    ValidatableToolProtocol, # + validate
)

---

Real-World Scenarios

Scenario 1: Unit Testing Agent Workflows

import pytest
from praisonaiagents.agent.protocols import AgentProtocol

class MockAgent:
    name = "mock"
    def chat(self, prompt, **kw): return "approved"
    async def achat(self, prompt, **kw): return "approved"

def test_approval_flow():
    """Test without LLM - fast, free, deterministic."""
    agent = MockAgent()
    result = agent.chat("Should I approve this?")
    assert result == "approved"

Scenario 2: Custom Database Memory

from praisonaiagents.memory.protocols import MemoryProtocol

class PostgresMemory:
    """Use PostgreSQL for enterprise deployments."""
    
    def __init__(self, conn_string):
        self.conn = psycopg2.connect(conn_string)
    
    def store_short_term(self, text, metadata=None, **kwargs):
        # Store in PostgreSQL
        pass
    
    def search_short_term(self, query, limit=5, **kwargs):
        # Vector search with pgvector
        pass
    
    # ... implement other methods

Scenario 3: Mock External API Tools

from praisonaiagents.tools.protocols import ToolProtocol

class MockWeatherTool:
    """Avoid rate limits in tests."""
    name = "weather"
    description = "Get weather"
    
    def run(self, city: str = "", **kwargs):
        return {"temp": 72, "condition": "sunny"}
    
    def get_schema(self):
        return {"type": "function", "function": {"name": "weather"}}

# Integration test without hitting real API
def test_weather_agent():
    agent = Agent(tools=[MockWeatherTool()])
    result = agent.chat("What's the weather?")
    assert "sunny" in result or "72" in result

---

Import Paths

# All protocols
from praisonaiagents.agent.protocols import AgentProtocol
from praisonaiagents.memory.protocols import MemoryProtocol
from praisonaiagents.tools.protocols import ToolProtocol

# From submodules directly
from praisonaiagents.agent import AgentProtocol
from praisonaiagents.memory import MemoryProtocol

---

Architecture: Where to Put Custom Implementations

Custom implementations (Redis, PostgreSQL, etc.) should NOT go in the core SDK. Use protocols to keep the core lightweight.

┌─────────────────────────────────────────────────────────────┐
│  praisonaiagents (CORE SDK)                                 │
│  • Lightweight, protocol-first                              │
│  • NO heavy dependencies (no redis, no postgres)            │
│  • Only: Protocol interfaces + lightweight defaults         │
│  • Fast import time                                         │
└─────────────────────────────────────────────────────────────┘
                          ▼
┌─────────────────────────────────────────────────────────────┐
│  praisonai (WRAPPER) or praisonai-tools                     │
│  • Heavy implementations live here                          │
│  • RedisMemory, PostgresMemory, MongoMemory                 │
│  • Optional deps: pip install praisonai[redis]              │
│  • Lazy imports (only loaded when used)                     │
└─────────────────────────────────────────────────────────────┘

Example: Adding Redis Memory

# FILE: praisonai-tools/praisonai_tools/memory/redis_memory.py
# NOT in praisonaiagents (core SDK)

from praisonaiagents.memory.protocols import MemoryProtocol

class RedisMemory:  # Implements MemoryProtocol
    """Redis-backed memory - lives in tools, NOT in core."""
    
    def __init__(self, redis_url: str):
        import redis  # Lazy import
        self.client = redis.from_url(redis_url)
    
    def store_short_term(self, text, metadata=None, **kwargs):
        key = f"stm:{hash(text)}"
        self.client.setex(key, 3600, text)
        return key
    
    def search_short_term(self, query, limit=5, **kwargs):
        return []
    
    def store_long_term(self, text, metadata=None, **kwargs):
        key = f"ltm:{hash(text)}"
        self.client.set(key, text)
        return key
    
    def search_long_term(self, query, limit=5, **kwargs):
        return []

Usage

from praisonai_tools.memory import RedisMemory
from praisonaiagents import Agent

agent = Agent(
    name="Redis Agent",
    memory=RedisMemory("redis://localhost:6379")
)

Benefits

Benefit	Explanation
Core stays fast	No redis import unless you use RedisMemory
Optional deps	pip install praisonai-tools[redis]
No bloat	Users who don’t need Redis don’t pay for it
Community extensions	Anyone can create compatible backends
Easy swapping	Same Agent code, just change memory= parameter

---

When to Update Protocols

Agent Change	Update Protocol?
Add new method	❌ No
Add new parameter	❌ No
Rename/remove chat/achat/name	⚠️ Yes (breaking)

Protocols define a minimal contract. Adding features to Agent doesn’t require protocol updates.