Architecture

SAM Framework Architecture

🏗️ Overview

SAM Framework implements a sophisticated event-driven architecture for AI-powered Solana blockchain operations. The framework combines modular design patterns with async-first principles to deliver high-performance, secure, and extensible agent capabilities.

Event-Driven Core

Async pub/sub system with real-time event streaming and component decoupling.

Plugin Architecture

Extensible tool system with entry point discovery and SDK integration.

Security First

Encrypted key storage, transaction validation, and comprehensive safety controls.

High Performance

uvloop optimization, connection pooling, and efficient RPC management.

Persistent Memory

SQLite-based conversation persistence with context compression.

Configurable Middleware

Pluggable pipeline for logging, rate limiting, and request processing.

🧩 Core Components

1. Event System (sam/core/events.py)

Central nervous system providing async pub/sub messaging between components.

class EventBus:
    async def publish(self, event: str, payload: Dict) -> None:
        # Async event publishing

    def subscribe(self, event: str, handler: Callable) -> None:
        # Event subscription management

Key Events:

  • tool.called - Tool execution initiated
  • tool.succeeded - Tool completed successfully
  • tool.failed - Tool execution failed
  • llm.usage - Token usage tracking
  • agent.session.started - Session lifecycle

2. Agent Engine (sam/core/agent.py)

Orchestrates AI conversation flow with advanced tool calling and loop prevention.

class SAMAgent:
    def __init__(
        self,
        llm: LLMProvider,
        tools: ToolRegistry,
        memory: MemoryManager,
        system_prompt: str,
        event_bus: EventBus
    ):
        # Full initialization with event integration

    async def run(self, user_input: str, session_id: str) -> str:
        # Advanced execution loop with tool loop prevention

Features:

  • Loop Prevention: Signature-based detection of repetitive tool calls
  • Error Categorization: Structured error handling with recovery strategies
  • Session Tracking: Context length and token usage monitoring
  • Tool Callbacks: Real-time execution status for CLI feedback

3. LLM Provider System (sam/core/llm_provider.py)

Multi-provider abstraction supporting OpenAI, Anthropic, xAI, and local models.

class LLMProvider(ABC):
    @abstractmethod
    async def chat_completion(
        self,
        messages: List[Dict],
        tools: List[Dict] = None
    ) -> ChatResponse:
        # Provider-specific implementation

class LLMProviderFactory:
    @staticmethod
    def create(provider_name: str) -> LLMProvider:
        # Factory pattern for provider instantiation

Supported Providers:

  • OpenAI (GPT models)
  • Anthropic (Claude models)
  • xAI (Grok models)
  • OpenAI-compatible (Ollama, LM Studio, vLLM)
  • Local API endpoints

4. Tool Registry (sam/core/tools.py)

Centralized tool management with middleware pipeline and plugin support.

class ToolRegistry:
    def __init__(self, middlewares: List[Middleware]):
        self._middlewares = middlewares
        self._tools = {}

    async def call(
        self,
        name: str,
        args: Dict,
        context: ToolContext
    ) -> Dict:
        # Middleware pipeline execution

Middleware Types:

  • Logging: Request/response tracking with configurable detail levels
  • Rate Limiting: Per-tool throttling with identifier-based limits
  • Retry: Exponential backoff for transient failures
  • Validation: Input sanitization and safety checks

5. Memory Manager (sam/core/memory.py)

Advanced SQLite-based persistent storage with context compression and session management.

class MemoryManager:
    async def save_session(self, session_id: str, messages: List[Dict]):
        # Persist conversation with compression

    async def load_session(self, session_id: str) -> List[Dict]:
        # Retrieve and decompress conversation history

    async def compact_conversation(self, session_id: str) -> str:
        # Compress old messages to maintain context window

Features:

  • Context Compression: Automatic summarization for long conversations
  • Session Isolation: Independent contexts per session ID
  • Efficient Storage: Optimized SQLite with indexing and WAL mode
  • Migration Support: Automatic schema updates and data migration

6. Agent Builder (sam/core/builder.py)

Factory pattern implementation for modular agent construction and dependency injection.

class AgentBuilder:
    def __init__(self):
        self._llm_provider = None
        self._tool_registry = None
        self._memory_manager = None
        self._middlewares = []

    def with_llm_provider(self, provider: LLMProvider) -> 'AgentBuilder':
        # Fluent interface for configuration

    async def build(self) -> SAMAgent:
        # Async construction with validation

Benefits:

  • Dependency Injection: Clean separation of concerns
  • Testability: Easy mocking for unit tests
  • Configuration Flexibility: Runtime configuration changes
  • Resource Management: Proper async cleanup coordination

🔧 Integration Layer

Solana Tools (sam/integrations/solana/)

Native Solana blockchain operations with comprehensive RPC integration.

class SolanaTools:
    def __init__(self, rpc_url: str, private_key: Optional[str] = None):
        # Initialize with secure key management

    async def get_balance(self, address: Optional[str] = None) -> Dict[str, Any]:
        # Complete wallet overview with token balances

Capabilities:

  • Wallet Operations: SOL transfers and balance checking
  • Token Management: SPL token account enumeration
  • Transaction Signing: Secure private key operations
  • RPC Optimization: Connection pooling and request caching

Trading Integrations

Pump.fun Tools (sam/integrations/pump_fun.py)

Real-time memecoin trading on the Pump.fun platform.

class PumpFunTools:
    async def pump_fun_buy(self, mint: str, amount: float, slippage: int) -> Dict[str, Any]:
        # Execute Pump.fun token purchases

Jupiter Tools (sam/integrations/jupiter.py)

DEX aggregation for optimal Solana token swaps.

class JupiterTools:
    async def get_swap_quote(self, input_mint: str, output_mint: str, amount: int) -> Dict[str, Any]:
        # Get optimized swap quotes across multiple DEXes

Market Data (sam/integrations/dexscreener.py)

Real-time market data and trading pair information.

class DexScreenerTools:
    async def search_pairs(self, query: str) -> Dict[str, Any]:
        # Search trading pairs by token or query

Web Search (sam/integrations/search.py)

Internet search capabilities via Brave Search API.

class SearchTools:
    async def search_web(self, query: str) -> Dict[str, Any]:
        # Web search with result summarization

🔌 Plugin System

Entry Point Discovery

SAM supports external tools and components via Python entry points for maximum extensibility.

# setup.py or pyproject.toml
[project.entry-points."sam.plugins"]
my_trading_tools = "my_package.tools:register"

[project.entry-points."sam.llm_providers"]
custom_llm = "my_package.llm:create_provider"

[project.entry-points."sam.memory_backends"]
redis_memory = "my_package.memory:create_backend"

[project.entry-points."sam.secure_storage"]
cloud_keyring = "my_package.security:create_storage"

Plugin Registration

# my_package/tools.py
from sam.core.tools import Tool, ToolRegistry

async def my_custom_tool(args: dict) -> dict:
    # Custom tool implementation
    return {"result": "custom logic executed"}

def register(registry: ToolRegistry, agent=None):
    """Register custom tools with the SAM agent."""
    registry.register(Tool(
        name="my_custom_tool",
        description="Custom trading tool",
        input_schema={
            "type": "object",
            "properties": {
                "param": {"type": "string", "description": "Tool parameter"}
            },
            "required": ["param"]
        },
        handler=my_custom_tool
    ))

Custom LLM Providers

# my_package/llm.py
from sam.core.llm_provider import LLMProvider
from sam.core.settings import Settings

class CustomLLMProvider(LLMProvider):
    def __init__(self, api_key: str, model: str):
        self.api_key = api_key
        self.model = model

    async def chat_completion(self, messages, tools=None):
        # Custom LLM implementation
        return ChatResponse(...)

def create_provider(settings: Settings = None) -> LLMProvider:
    """Factory function for custom LLM provider."""
    return CustomLLMProvider(
        api_key=settings.CUSTOM_API_KEY,
        model=settings.CUSTOM_MODEL
    )

Environment Variables

# Load custom plugins
export SAM_PLUGINS="my_package.tools,vendor_package.tools"

# Use custom memory backend
export SAM_MEMORY_BACKEND="my_package.memory:create_backend"

# Custom middleware configuration
export SAM_MIDDLEWARE_JSON='{"logging": {"include_args": true}}'

🛡️ Security Architecture

Key Management

OS Keyring Integration

# sam/utils/secure_storage.py
class SecureStorage:
    def store_private_key(self, key_id: str, private_key: str):
        # Store encrypted keys in system keyring

    def get_private_key(self, key_id: str) -> Optional[str]:
        # Retrieve and decrypt keys securely

Features:

  • System-level credential storage
  • Automatic key retrieval
  • Cross-platform compatibility
  • Secure key lifecycle management

⚡ Performance Optimizations

Async Architecture

  • uvloop Integration: High-performance event loop
  • Connection Pooling: Reusable HTTP connections
  • Concurrent Requests: Parallel tool execution where possible
  • Smart Caching: Session-based result caching

Resource Management

  • Automatic Cleanup: Proper resource disposal
  • Memory Monitoring: Leak prevention and optimization
  • Rate Limiting: Configurable request throttling
  • Error Recovery: Graceful handling of network issues

📊 Data Flow

graph TD
    A[User Input] --> B[SAM Agent]
    B --> C[LLM Provider]
    C --> D{Tool Call?}

    D -->|Yes| E[Tool Registry]
    D -->|No| F[Final Response]

    E --> G[Execute Tool]
    G --> H[Tool Result]
    H --> C

    F --> I[Memory Manager]
    I --> J[Persist Context]

    G --> K[Error Handler]
    K --> L[Graceful Degradation]

🔧 Configuration System

Environment-Based Config

# sam/config/settings.py
class Settings:
    # Centralized configuration management
    OPENAI_API_KEY: str
    SAM_FERNET_KEY: Optional[str]
    # ... other settings

Dynamic Loading

  • Environment variable parsing
  • Validation and type checking
  • Runtime configuration updates
  • Secure credential management

🧪 Testing Architecture

Comprehensive Test Suite

  • Unit Tests: Individual component testing
  • Integration Tests: End-to-end workflow validation
  • Security Tests: Key management and encryption validation
  • Performance Tests: Load testing and optimization

Test Structure

tests/
├── test_tools.py          # Tool functionality
├── test_integration.py    # Cross-component tests
├── test_crypto.py         # Security features
├── test_memory.py         # Persistence layer
└── test_rate_limiter.py   # Performance controls

🚀 Deployment Considerations

Production Setup

Monitoring and Observability

  • Health Checks: System status monitoring
  • Usage Analytics: Token consumption tracking
  • Error Reporting: Comprehensive error logging
  • Performance Metrics: Response time and throughput monitoring

🏗️ Architecture Principles

  • • Modular Design: Clean separation of concerns with plugin-based tools
  • • Security First: Defense-in-depth with multiple security layers
  • • Performance Optimized: Async-first with efficient resource usage
  • • Developer Friendly: Comprehensive testing and clear APIs
  • • Production Ready: Robust error handling and monitoring