ipckit

A cross-platform IPC (Inter-Process Communication) library for Rust and Python, implemented in Rust.

Features

Performance - Implemented in Rust; uses zero-copy where possible
Cross-Platform - Runs on Windows, Linux, and macOS
Python Bindings - Python support via PyO3
Multiple IPC Methods - Pipes, shared memory, channels, and file-based IPC
Thread Safety - Safe concurrent access across processes
Native JSON - JSON serialization using Rust's serde_json
Graceful Shutdown - Support for graceful channel shutdown
Local Socket - Unix Domain Socket / Named Pipe abstraction for socket communication
Thread Channel - A channel for communication between threads in the same process
Event Stream - Publish-subscribe event system
Task Manager - Task lifecycle management with progress tracking
Socket Server - Multi-client socket server (like Docker's socket)
CLI Bridge - Integrate CLI tools with progress reporting and communication
Channel Metrics - Metrics tracking for send/receive operations
CLI Tools - Code generation and channel monitoring commands
Declarative Macros - Macros for channel creation and command routing

Installation

Python

pip install ipckit

Prebuilt wheels are published per platform:

cp38-abi3 wheels for CPython 3.8 and newer (linux x86_64, linux aarch64, win_amd64, macOS universal2).
cp37-cp37m wheels for CPython 3.7 (linux x86_64, win_amd64).
py3-none-any "py37-lite" wheel (type stubs only, no native module).

Maya 2022 and Blender 2.83 embed CPython 3.7, which cannot load cp38-abi3 wheels. Those hosts use the native cp37-cp37m wheel. On other CPython 3.7 platforms, the py37-lite wheel provides type stubs for static analysis but no native module.

Rust

[dependencies]
ipckit = "0.1"

Quick Start

Anonymous Pipe (Parent-Child Communication)

Python:

import ipckit
import subprocess

# Create pipe pair
pipe = ipckit.AnonymousPipe()

# Write to pipe
pipe.write(b"Hello from parent!")

# Read from pipe
data = pipe.read(1024)
print(data)

Rust:

use ipckit::AnonymousPipe;

fn main() -> ipckit::Result<()> {
    let pipe = AnonymousPipe::new()?;
    
    pipe.write_all(b"Hello from Rust!")?;
    
    let mut buf = [0u8; 1024];
    let n = pipe.read(&mut buf)?;
    println!("{}", String::from_utf8_lossy(&buf[..n]));
    
    Ok(())
}

Named Pipe (Unrelated Process Communication)

Python Server:

import ipckit

# Create server
server = ipckit.NamedPipe.create("my_pipe")
print("Waiting for client...")
server.wait_for_client()

# Communicate
data = server.read(1024)
server.write(b"Response from server")

Python Client:

import ipckit

# Connect to server
client = ipckit.NamedPipe.connect("my_pipe")

# Communicate
client.write(b"Hello from client")
response = client.read(1024)
print(response)

Shared Memory (Fast Data Exchange)

Python:

import ipckit

# Create shared memory (owner)
shm = ipckit.SharedMemory.create("my_shm", 4096)
shm.write(0, b"Shared data here!")

# In another process, open existing
shm2 = ipckit.SharedMemory.open("my_shm")
data = shm2.read(0, 17)
print(data)  # b"Shared data here!"

Rust:

use ipckit::SharedMemory;

fn main() -> ipckit::Result<()> {
    // Create
    let shm = SharedMemory::create("my_shm", 4096)?;
    shm.write(0, b"Hello from Rust!")?;
    
    // Open in another process
    let shm2 = SharedMemory::open("my_shm")?;
    let data = shm2.read(0, 16)?;
    
    Ok(())
}

IPC Channel (High-Level Message Passing)

Python:

import ipckit

# Server
channel = ipckit.IpcChannel.create("my_channel")
channel.wait_for_client()

# Send/receive JSON
channel.send_json({"type": "greeting", "message": "Hello!"})
response = channel.recv_json()
print(response)

File Channel (Frontend-Backend Communication)

For desktop applications where a Python backend communicates with a web frontend.

Python Backend:

import ipckit

# Create backend channel
channel = ipckit.FileChannel.backend("./ipc_channel")

# Send request to frontend
request_id = channel.send_request("getData", {"key": "user_info"})

# Wait for response
response = channel.wait_response(request_id, timeout_ms=5000)
print(response)

# Send events
channel.send_event("status_update", {"status": "ready"})

JavaScript Frontend:

// Read from: ./ipc_channel/backend_to_frontend.json
// Write to:  ./ipc_channel/frontend_to_backend.json

async function pollMessages() {
    const response = await fetch('./ipc_channel/backend_to_frontend.json');
    const messages = await response.json();
    // Process new messages...
}

Native JSON Functions

ipckit provides Rust-native JSON functions backed by Rust's serde_json:

import ipckit

# Serialize
data = {"name": "test", "values": [1, 2, 3]}
json_str = ipckit.json_dumps(data)

# Pretty print
pretty_str = ipckit.json_dumps_pretty(data)

# Deserialize
obj = ipckit.json_loads('{"key": "value"}')

Graceful Shutdown

When using IPC channels with event loops (like WebView, GUI frameworks), background threads may continue sending messages after the main event loop has closed, causing errors. The GracefulChannel feature solves this problem.

Python:

import ipckit

# Create channel with graceful shutdown support
channel = ipckit.GracefulIpcChannel.create("my_channel")
channel.wait_for_client()

# ... use channel normally ...
data = channel.recv()
channel.send(b"response")

# Graceful shutdown - prevents new operations and waits for pending ones
channel.shutdown()
channel.drain()  # Wait for all pending operations to complete

# Or use shutdown with timeout (in milliseconds)
channel.shutdown_timeout(5000)  # 5 second timeout

Rust:

use ipckit::{GracefulIpcChannel, GracefulChannel};
use std::time::Duration;

fn main() -> ipckit::Result<()> {
    let mut channel = GracefulIpcChannel::<Vec<u8>>::create("my_channel")?;
    channel.wait_for_client()?;
    
    // ... use channel ...
    
    // Graceful shutdown
    channel.shutdown();
    channel.drain()?;
    
    // Or with timeout
    channel.shutdown_timeout(Duration::from_secs(5))?;
    
    Ok(())
}

Key Benefits:

Prevents EventLoopClosed and similar errors
Shutdown signaling is thread-safe
Tracks pending operations with RAII guards
Configurable drain timeout

Local Socket (Cross-Platform Socket Communication)

Local sockets provide a unified API for Unix Domain Sockets (Unix/macOS) and Named Pipes (Windows).

Python Server:

import ipckit

# Create server
server = ipckit.LocalSocketListener.bind("my_socket")
print("Waiting for client...")

# Accept connection
stream = server.accept()

# Receive and send data
data = stream.read(1024)
print(f"Received: {data}")
stream.write(b"Hello from server!")

# JSON communication
json_data = stream.recv_json()
stream.send_json({"status": "ok", "message": "received"})

Python Client:

import ipckit

# Connect to server
stream = ipckit.LocalSocketStream.connect("my_socket")

# Send and receive data
stream.write(b"Hello from client!")
response = stream.read(1024)
print(f"Response: {response}")

# JSON communication
stream.send_json({"action": "getData", "key": "user"})
result = stream.recv_json()
print(result)

Key Benefits:

Cross-platform: Runs on Windows, Linux, and macOS
Bidirectional communication
JSON serialization with length prefix
Simple client-server model

Thread Channel (Intra-Process Communication)

A channel for communication between threads within the same process.

Rust:

use ipckit::ThreadChannel;
use std::thread;

fn main() {
    // Create an unbounded channel
    let (tx, rx) = ThreadChannel::<String>::unbounded();

    // Spawn producer thread
    let tx_clone = tx.clone();
    thread::spawn(move || {
        tx_clone.send("Hello from thread!".to_string()).unwrap();
    });

    // Receive in main thread
    let msg = rx.recv().unwrap();
    println!("Received: {}", msg);
}

Event Stream (Publish-Subscribe)

Event system for task progress, logs, and notifications.

Python:

import ipckit

# Create event bus
bus = ipckit.EventBus()
publisher = bus.publisher()

# Subscribe to task events
subscriber = bus.subscribe(ipckit.EventFilter().event_type("task.*"))

# Publish events
publisher.progress("task-123", 50, 100, "Half done")
publisher.log("task-123", "info", "Processing...")

# Receive events (non-blocking)
while event := subscriber.try_recv():
    print(f"[{event.event_type}] {event.data}")

# Or with timeout
try:
    event = subscriber.recv_timeout(1000)  # 1 second
except RuntimeError:
    print("Timeout")

Rust:

use ipckit::{EventBus, Event, EventFilter};

fn main() {
    let bus = EventBus::new(Default::default());
    let publisher = bus.publisher();

    // Subscribe to task events
    let subscriber = bus.subscribe(
        EventFilter::new().event_type("task.*")
    );

    // Publish events
    publisher.progress("task-123", 50, 100, "Half done");
    publisher.log("task-123", "info", "Processing...");

    // Receive events
    while let Some(event) = subscriber.try_recv() {
        println!("[{}] {:?}", event.event_type, event.data);
    }
}

Task Manager (Task Lifecycle)

Manage long-running tasks with progress tracking and cancellation support.

Python:

import ipckit
import time

manager = ipckit.TaskManager()

# Create a task
handle = manager.create_task("Upload files", "upload")
handle.start()

# Simulate work
for i in range(100):
    if handle.is_cancelled:
        handle.fail("Cancelled by user")
        break
    handle.set_progress(i + 1, f"Step {i + 1}/100")
    time.sleep(0.01)
else:
    handle.complete({"uploaded": 100})

# List active tasks
active = manager.list_active()
print(f"Active tasks: {len(active)}")

# Cancel a task
# manager.cancel(handle.id)

Rust:

use ipckit::{TaskManager, TaskBuilder, TaskFilter};
use std::time::Duration;

fn main() {
    let manager = TaskManager::new(Default::default());

    // Spawn a task
    let handle = manager.spawn("Upload files", "upload", |task| {
        for i in 0..100 {
            if task.is_cancelled() {
                return;
            }
            task.set_progress(i + 1, Some(&format!("Step {}/100", i + 1)));
            std::thread::sleep(Duration::from_millis(50));
        }
        task.complete(serde_json::json!({"uploaded": 100}));
    });

    // List active tasks
    let active = manager.list(&TaskFilter::new().active());
    println!("Active tasks: {}", active.len());

    // Cancel if needed
    // manager.cancel(handle.id()).unwrap();
}

Socket Server (Multi-Client Server)

Docker-style socket server for handling multiple client connections.

Rust:

use ipckit::{SocketServer, SocketServerConfig, Message, FnHandler};

fn main() -> ipckit::Result<()> {
    let server = SocketServer::new(SocketServerConfig::with_path("my_server"))?;

    // Handle connections with a simple function
    let handler = FnHandler::new(|conn, msg| {
        if msg.method() == Some("ping") {
            Ok(Some(Message::response(serde_json::json!({"pong": true}))))
        } else {
            Ok(None)
        }
    });

    // Run server (blocking)
    server.run(handler)?;
    Ok(())
}

Client:

use ipckit::SocketClient;

fn main() -> ipckit::Result<()> {
    let mut client = SocketClient::connect("my_server")?;

    // Send request and get response
    let result = client.request("ping", serde_json::json!({}))?;
    println!("Response: {:?}", result);

    Ok(())
}

API Server (HTTP-style API over Local Socket)

For Python server-side applications, you can integrate with async frameworks such as FastAPI or Robyn. These frameworks provide routing, middleware, and async support.

Python with FastAPI + Uvicorn (Unix Socket):

# server.py
from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/v1/health")
async def health():
    return {"status": "ok"}

@app.post("/v1/tasks")
async def create_task(data: dict):
    return {"id": "task-123", "name": data.get("name")}

# Run on Unix socket
if __name__ == "__main__":
    uvicorn.run(app, uds="/tmp/my_api.sock")

Python with Robyn:

# server.py
from robyn import Robyn

app = Robyn(__file__)

@app.get("/v1/health")
async def health():
    return {"status": "ok"}

@app.post("/v1/tasks")
async def create_task(request):
    data = request.json()
    return {"id": "task-123", "name": data.get("name")}

# Robyn supports Unix sockets via configuration
app.start(host="0.0.0.0", port=8080)

Python Client (using ipckit):

import ipckit

# Connect to the API server
client = ipckit.ApiClient("/tmp/my_api.sock")

# Make requests
health = client.get("/v1/health")
print(health)  # {"status": "ok"}

task = client.post("/v1/tasks", {"name": "my-task"})
print(task)  # {"id": "task-123", "name": "my-task"}

Rust Server:

use ipckit::{ApiServer, ApiServerConfig, Router, Response};

fn main() -> ipckit::Result<()> {
    let config = ApiServerConfig::new("/tmp/my_api.sock");
    
    let router = Router::new()
        .get("/v1/health", |_req| {
            Response::ok(serde_json::json!({"status": "ok"}))
        })
        .post("/v1/tasks", |req| {
            let data = req.json::<serde_json::Value>()?;
            Response::created(serde_json::json!({
                "id": "task-123",
                "name": data.get("name")
            }))
        });
    
    let server = ApiServer::new(config, router)?;
    server.run()?;
    Ok(())
}

CLI Bridge (CLI Tool Integration)

Integrate a CLI tool with progress tracking and bidirectional communication.

Python:

import ipckit

# Method 1: Use CliBridge directly
bridge = ipckit.CliBridge()
bridge.register_task("Build Project", "build")

for i in range(100):
    if bridge.is_cancelled:
        bridge.fail("Cancelled by user")
        break
    bridge.set_progress(i + 1, f"Step {i + 1}/100")

bridge.complete({"success": True})

# Method 2: Wrap existing command with progress parsing
output = ipckit.wrap_command(
    ["cargo", "build", "--release"],
    task_name="Build Project",
    task_type="build"
)
print(f"Exit code: {output.exit_code}")
print(f"Duration: {output.duration_ms}ms")

# Method 3: Parse progress from output
info = ipckit.parse_progress("Downloading... 75%", "percentage")
print(f"Progress: {info.percentage}%")

Rust:

use ipckit::{CliBridge, WrappedCommand, parsers};

fn main() -> ipckit::Result<()> {
    // Method 1: Direct bridge usage
    let bridge = CliBridge::connect()?;
    bridge.register_task("My Task", "build")?;
    
    for i in 0..100 {
        if bridge.is_cancelled() {
            bridge.fail("Cancelled");
            return Ok(());
        }
        bridge.set_progress(i + 1, Some(&format!("Step {}/100", i + 1)));
    }
    bridge.complete(serde_json::json!({"success": true}));

    // Method 2: Wrap existing command
    let output = WrappedCommand::new("cargo")
        .args(["build", "--release"])
        .task("Build Project", "build")
        .progress_parser(parsers::PercentageParser)
        .run()?;
    
    println!("Exit code: {}", output.exit_code);
    Ok(())
}

Key Features:

Automatic stdout/stderr capture and forwarding
Progress parsers for percentage, fraction, and progress bar
Task cancellation support
Minimal invasiveness - existing CLI needs minimal modifications

Channel Metrics (Performance Monitoring)

Track send/receive operations with metrics.

Rust:

use ipckit::{ChannelMetrics, MeteredSender, MeteredReceiver, metered_pair, AggregatedMetrics};
use std::sync::Arc;

fn main() {
    // Create metered sender/receiver pair
    let (tx, rx) = metered_pair(crossbeam_channel::unbounded());
    
    // Send messages
    tx.send("Hello".to_string()).unwrap();
    tx.send("World".to_string()).unwrap();
    
    // Receive messages
    let _ = rx.recv().unwrap();
    
    // Get metrics
    let metrics = tx.metrics();
    println!("Sent: {}, Received: {}", metrics.messages_sent(), metrics.messages_received());
    
    // Aggregate metrics from multiple channels
    let mut aggregated = AggregatedMetrics::new();
    aggregated.add_channel("channel1", metrics.clone());
    
    // Export as JSON or Prometheus format
    println!("{}", aggregated.to_json());
    println!("{}", aggregated.to_prometheus());
}

CLI Tools

ipckit provides CLI tools for code generation and channel monitoring.

Code Generation:

# Generate client code
ipckit generate client --name MyClient --output ./src/client.rs

# Generate server code
ipckit generate server --name MyServer --output ./src/server.rs

# Generate Python bindings
ipckit generate python --name my_module --output ./bindings/

# Generate message handler
ipckit generate handler --name MessageHandler --output ./src/handler.rs

Channel Monitoring:

# Monitor channel with TUI interface
ipckit monitor --channel my_channel

# Monitor with JSON output
ipckit monitor --channel my_channel --format json

# Monitor with custom refresh interval
ipckit monitor --channel my_channel --interval 500

Declarative Macros

Macros for common IPC patterns.

Rust:

use ipckit::{ipc_channel, ipc_commands, ipc_message, ipc_middleware};

fn main() {
    // Create a channel with a single macro
    let (tx, rx) = ipc_channel!(String, "my_channel");
    
    // Define message types
    ipc_message! {
        struct UserRequest {
            user_id: u64,
            action: String,
        }
    }
    
    // Define command routing
    ipc_commands! {
        "ping" => handle_ping,
        "echo" => handle_echo,
        "status" => handle_status,
    }
    
    // Chain middleware
    ipc_middleware! {
        logging_middleware,
        auth_middleware,
        => final_handler
    }
}

fn handle_ping() -> String { "pong".to_string() }
fn handle_echo() -> String { "echo".to_string() }
fn handle_status() -> String { "ok".to_string() }
fn logging_middleware<F: Fn() -> String>(next: F) -> String { next() }
fn auth_middleware<F: Fn() -> String>(next: F) -> String { next() }
fn final_handler() -> String { "done".to_string() }

IPC Methods Comparison

Method	Use Case	Performance	Complexity
Anonymous Pipe	Parent-child processes	High	Low
Named Pipe	Unrelated processes	High	Medium
Shared Memory	Large data, frequent access	Very high	High
IPC Channel	Message passing	High	Low
File Channel	Frontend-backend	Medium	Low
Graceful Channel	Event loop integration	High	Low
Local Socket	Cross-platform sockets	High	Low
Thread Channel	Intra-process threads	Very high	Low
Event Stream	Publish-subscribe events	High	Low
Task Manager	Task lifecycle	High	Medium
Socket Server	Multi-client server	High	Medium
CLI Bridge	CLI tool integration	High	Low
Channel Metrics	Performance monitoring	High	Low
CLI Tools	Code generation & monitoring	N/A	Low
Declarative Macros	Boilerplate reduction	N/A	Low

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Python Application                       │
├─────────────────────────────────────────────────────────────┤
│                    ipckit Python Bindings                    │
│                         (PyO3)                               │
├─────────────────────────────────────────────────────────────┤
│                     ipckit Rust Core                         │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────────────────┐│
│  │  Pipes  │ │   SHM   │ │ Channel │ │    File Channel     ││
│  └─────────┘ └─────────┘ └─────────┘ └─────────────────────┘│
│  ┌─────────────────────────────────────────────────────────┐│
│  │              Graceful Shutdown Layer                    ││
│  │  (GracefulNamedPipe, GracefulIpcChannel, ShutdownState) ││
│  └─────────────────────────────────────────────────────────┘│
│  ┌─────────────────────────────────────────────────────────┐│
│  │                  Local Socket Layer                     ││
│  │     (LocalSocketListener, LocalSocketStream)            ││
│  └─────────────────────────────────────────────────────────┘│
│  ┌─────────────────────────────────────────────────────────┐│
│  │                  High-Level Services                    ││
│  │  (ThreadChannel, EventStream, TaskManager, SocketServer)││
│  │  (CliBridge, WrappedCommand)                            ││
│  └─────────────────────────────────────────────────────────┘│
├─────────────────────────────────────────────────────────────┤
│              Platform Abstraction Layer                      │
│         (Windows / Linux / macOS)                            │
└─────────────────────────────────────────────────────────────┘

Building from Source

Prerequisites

Rust 1.70+
Python 3.7+
maturin (pip install maturin)

Build

# Clone repository
git clone https://github.com/loonghao/ipckit.git
cd ipckit

# Build Python package
maturin develop --release

# Run tests
pytest tests/
cargo test

License

This project is dual-licensed under:

Contributing

Contributions are welcome. Please submit a Pull Request.

Fork the repository
Create your feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add some amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Documentation

Acknowledgments

PyO3 - Rust bindings for Python
maturin - Build and publish Rust-based Python packages
serde - Serialization framework for Rust

Name		Name	Last commit message	Last commit date
.agents/skills		.agents/skills
.claude/skills		.claude/skills
.cline/skills		.cline/skills
.codebuddy/skills		.codebuddy/skills
.cursor/skills		.cursor/skills
.github		.github
.kiro/skills		.kiro/skills
.opencode/skills		.opencode/skills
.roo/skills		.roo/skills
.trae/skills		.trae/skills
.windsurf/skills		.windsurf/skills
crates		crates
examples		examples
python/ipckit		python/ipckit
scripts		scripts
tests		tests
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
.release-please-manifest.json		.release-please-manifest.json
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
Cargo.toml		Cargo.toml
LICENSE-APACHE		LICENSE-APACHE
LICENSE-MIT		LICENSE-MIT
README.md		README.md
README_zh.md		README_zh.md
pyproject.toml		pyproject.toml
release-please-config.json		release-please-config.json
renovate.json		renovate.json

Navigation Menu

Repository files navigation

ipckit

Features

Installation

Python

Rust

Quick Start

Anonymous Pipe (Parent-Child Communication)

Named Pipe (Unrelated Process Communication)

Shared Memory (Fast Data Exchange)

IPC Channel (High-Level Message Passing)

File Channel (Frontend-Backend Communication)

Native JSON Functions

Graceful Shutdown

Local Socket (Cross-Platform Socket Communication)

Thread Channel (Intra-Process Communication)

Event Stream (Publish-Subscribe)

Task Manager (Task Lifecycle)

Socket Server (Multi-Client Server)

API Server (HTTP-style API over Local Socket)

CLI Bridge (CLI Tool Integration)

Channel Metrics (Performance Monitoring)

CLI Tools

Declarative Macros

IPC Methods Comparison

Architecture

Building from Source

Prerequisites

Build

License

Contributing

Documentation

Acknowledgments

About

Releases

Packages

Used by

Contributors

Languages

Latest commit

History

Folders and files

Repository files navigation

ipckit

Features

Installation

Python

Rust

Quick Start

Anonymous Pipe (Parent-Child Communication)

Named Pipe (Unrelated Process Communication)

Shared Memory (Fast Data Exchange)

IPC Channel (High-Level Message Passing)

File Channel (Frontend-Backend Communication)

Native JSON Functions

Graceful Shutdown

Local Socket (Cross-Platform Socket Communication)

Thread Channel (Intra-Process Communication)

Event Stream (Publish-Subscribe)

Task Manager (Task Lifecycle)

Socket Server (Multi-Client Server)

API Server (HTTP-style API over Local Socket)

CLI Bridge (CLI Tool Integration)

Channel Metrics (Performance Monitoring)

CLI Tools

Declarative Macros

IPC Methods Comparison

Architecture

Building from Source

Prerequisites

Build

License

Contributing

Documentation

Acknowledgments

About

Resources

Contributing

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages