RPC - Redis-based RPC Framework

A Rust framework for building distributed systems using Redis for configuration, logging, error tracking, and RPC queues. Optionally includes MCP (Model Context Protocol) client support.

Features

• ConfigClient: Redis configuration management using OTOML format
• RedisLogger: Distributed logging to Redis
• ErrorStore: Redis-based error storage and tracking
• RpcQueue: Redis queue-based RPC mechanism
• McpClient: Connect to MCP servers via HTTP (optional, enabled by default)

Installation

[dependencies]
rpc = "0.1"

# Without MCP support:
# rpc = { version = "0.1", default-features = false }

Quick Start

Configuration Management

use rpc::config::{ConfigClient, LlmConfig, LlmProvider};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = ConfigClient::new("redis://localhost:6379").await?;

    // Get an LLM configuration
    let llm: LlmConfig = client.get_llm("openai", "default").await?;
    println!("Default model: {}", llm.default_model.unwrap_or_default());

    // List all LLM providers
    let providers = client.list_types("llm").await?;
    println!("Providers: {:?}", providers);

    // Set a configuration
    let config = LlmConfig {
        provider: LlmProvider::Openai,
        name: "default".to_string(),
        description: Some("OpenAI default config".to_string()),
        api_key_env: Some("OPENAI_API_KEY".to_string()),
        default_model: Some("gpt-4-turbo".to_string()),
        enabled: true,
        ..Default::default()
    };
    client.set_llm("openai", "default", &config).await?;

    Ok(())
}

Demo Data Generation

use rpc::config::{ConfigClient, generate_demodata};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = ConfigClient::new("redis://localhost:6379").await?;
    
    // Generate sample configurations for UI development
    let stats = generate_demodata(&client).await?;
    println!("Generated {} configurations", stats.total());

    Ok(())
}

Redis Logging

use rpc::logger::{RedisLogger, LogLevel};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let logger = RedisLogger::builder()
        .redis_url("redis://localhost:6379")
        .service("api")
        .environment("prod")
        .build()
        .await?;

    logger.info("Application started").await?;
    logger.warn("High memory usage").await?;
    logger.error("Database connection failed").await?;

    // Retrieve recent logs
    let logs = logger.get_logs(LogLevel::Error, 10).await?;
    for log in logs {
        println!("{}", log.to_log_line());
    }

    Ok(())
}

Error Storage

use rpc::errors::ErrorStore;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let store = ErrorStore::new("redis://localhost:6379", "api", "prod").await?;

    // Store an error
    let error_id = store.store_error(
        "database",
        "ConnectionError",
        "Failed to connect to database",
        Some("Full backtrace here...".to_string()),
    ).await?;

    // Retrieve errors by category
    let errors = store.get_errors("database", 10).await?;
    for err in errors {
        println!("[{}] {}: {}", err.timestamp, err.kind, err.message);
    }

    // Mark as resolved
    store.resolve_error(&error_id, Some("Fixed connection pool".to_string())).await?;

    Ok(())
}

RPC Queue

use rpc::rpc::{RpcQueue, RpcResponse};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let rpc = RpcQueue::new("redis://localhost:6379").await?
        .service_name("api_gateway");

    // Send a request
    let request_id = rpc.send_request(
        "user_service",
        "get_user",
        serde_json::json!({"id": 123}),
    ).await?;

    // Wait for response (with timeout)
    let response = rpc.wait_response(&request_id, 30).await?;
    
    if response.is_success() {
        println!("Result: {:?}", response.data);
    }

    Ok(())
}

RPC Worker

use rpc::rpc::{RpcQueue, RpcResponse};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let rpc = RpcQueue::new("redis://localhost:6379").await?;

    loop {
        // Wait for requests
        if let Some(request) = rpc.receive_request("user_service", "get_user", 5.0).await? {
            println!("Received request: {}", request.id);
            
            // Process and respond
            let result = serde_json::json!({"name": "Alice", "id": 123});
            rpc.respond_success(&request.id, result).await?;
        }
    }
}

MCP Client (Optional)

use rpc::mcp::McpClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = McpClient::builder()
        .endpoint("http://localhost:3000/mcp")
        .build()
        .await?;

    let tools = client.list_tools().await?;
    for tool in tools {
        println!("{}: {:?}", tool.name, tool.description);
    }

    let result = client.call_tool("echo", Some(serde_json::json!({
        "message": "Hello, world!"
    }))).await?;

    println!("{:?}", result);
    Ok(())
}

Configuration Categories

The ConfigClient supports these configuration categories (based on redis_config.md spec):

Key Structure

# Configuration
cfg:{category}:{type}:{instance} -> OTOML data

# Logging
log:{environment}:{service}:{level} -> List of log entries

# Errors
err:{environment}:{service}:{category} -> List of error IDs
err:entry:{error_id} -> Error entry JSON

# RPC
rpc:queue:{actor}:{method} -> Request queue
rpc:response:{request_id} -> Response data
rpc:pending:{actor} -> Pending request IDs

Project Structure

rpc/
├── src/
│   ├── lib.rs           # Library entry point
│   ├── config/          # Configuration management
│   │   ├── mod.rs
│   │   ├── client.rs    # ConfigClient implementation
│   │   ├── types.rs     # Config type definitions
│   │   ├── demodata.rs  # Demo data generation
│   │   └── error.rs
│   ├── logger/          # Redis logging
│   │   ├── mod.rs
│   │   ├── logger.rs    # RedisLogger implementation
│   │   ├── entry.rs     # LogEntry types
│   │   ├── subscriber.rs
│   │   └── error.rs
│   ├── errors/          # Error storage
│   │   ├── mod.rs
│   │   ├── store.rs     # ErrorStore implementation
│   │   ├── entry.rs     # ErrorEntry types
│   │   └── error.rs
│   ├── rpc/             # RPC queue
│   │   ├── mod.rs
│   │   ├── queue.rs     # RpcQueue implementation
│   │   ├── request.rs   # RpcRequest types
│   │   ├── response.rs  # RpcResponse types
│   │   └── error.rs
│   └── mcp/             # MCP client (optional)
│       ├── mod.rs
│       ├── mcpclient.rs
│       ├── rpcclient.rs
│       ├── rpcserver.rs
│       └── error.rs
├── build.sh
├── run.sh
├── install.sh
└── README.md

Requirements

• Rust 1.92.0 or later
• Edition 2024
• Redis server

Dependencies

• redis - Redis client
• herolib_osis - OTOML serialization
• tokio - Async runtime
• serde / serde_json - Serialization
• thiserror - Error handling
• rmcp - MCP client (optional)

Building

./build.sh

License

MIT