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