C-Two is a type-safe Remote Procedure Call (RPC) framework designed for distributed resource computation systems. The framework provides a structured abstraction layer that enables remote method invocation between client Components and Core Resource Models (CRM) with automatic serialization and protocol-agnostic communication.
C-Two addresses the complexity of distributed computing by implementing a clear separation of concerns through interface-based programming and automatic type marshalling. The framework enables developers to write distributed applications using familiar local programming patterns while maintaining type safety across network boundaries.
Unlike traditional stateless RPC frameworks, C-Two is specifically designed for resource-oriented computing, where computational resources maintain state and provide domain-specific operations. This approach is particularly valuable for geographic information systems, scientific computing, and other domains requiring persistent resource management with distributed access patterns.
- Resource-Oriented Design: Focus on stateful computational resources rather than stateless function calls
- Interface Segregation: Clean separation between interface definitions (ICRM) and implementations (CRM)
- Type Safety: Compile-time and runtime type checking with automatic or customized serialization inference
- Protocol Abstraction: Transport-agnostic communication supporting multiple protocols (Memory, IPC, TCP, HTTP, MCP)
- Resource Isolation: Encapsulation of computational resources behind well-defined interfaces
The framework implements a three-tier architecture:
Client-side computational units that consume remote resources through Interface of Core Resource Models (ICRM), providing network transparency for distributed operations.
Server-side Core Resource Models implementing domain-specific business logic and resource management, exposed through standardized ICRM interfaces. CRMs maintain persistent state and provide resource-specific operations.
Protocol-agnostic communication infrastructure supporting multiple transport mechanisms (Memory, IPC, TCP, HTTP) with automatic protocol detection, connection management and message serialization. The framework automatically selects the appropriate transport implementation based on the address scheme.
Install the latest stable version from PyPI using pip:
pip install c-twoFor better dependency resolution, use uv:
# Add c-two to your project
uv add c-twoFor development, testing, or accessing the latest features:
# Clone the repository
git clone https://github.com/world-in-progress/c-two.git
# Navigate to the project directory
cd c-two
# Install development dependencies
uv syncDefine remote interfaces using the @icrm decorator to establish contracts for distributed communication:
import c_two as cc
@cc.icrm
class IGrid:
def get_grid_infos(self, level: int, global_ids: list[int]) -> list[GridAttribute]:
"""Retrieve grid information for specified level and identifiers"""
...
def subdivide_grids(self, levels: list[int], global_ids: list[int]) -> list[str]:
"""Perform grid subdivision operations"""
...Implement the Core Resource Model using the @iicrm decorator:
@cc.iicrm
class Grid(IGrid):
def __init__(self, epsg: int, bounds: list, first_size: list[float], subdivide_rules: list[list[int]]):
self.epsg = epsg
self.bounds = bounds
# Resource initialization
def get_grid_infos(self, level: int, global_ids: list[int]) -> list[GridAttribute]:
# Business logic implementation with automatic serialization
return [GridAttribute(level=level, global_id=gid, ...) for gid in global_ids]
def subdivide_grids(self, levels: list[int], global_ids: list[int]) -> list[str]:
# Subdivision algorithm implementation
return [f"{level+1}-{child_id}" for level, gid in zip(levels, global_ids) for child_id in children]Define serializable data structures using the @transferable decorator with any necessary serialization logic:
import pyarrow as pa
@cc.transferable
class GridAttribute:
level: int
global_id: int
elevation: float
def serialize(data: 'GridAttribute') -> bytes:
schema = pa.schema([
pa.field('level', pa.int32()),
pa.field('global_id', pa.int32()),
pa.field('elevation', pa.float64())
])
table = pa.Table.from_pylist([data.__dict__], schema=schema)
return serialize_from_table(table)
def deserialize(arrow_bytes: bytes) -> 'GridAttribute':
row = deserialize_to_rows(arrow_bytes)[0]
return GridAttribute(**row)
@cc.transferable
class GridAttributes:
"""
A collection of GridAttribute objects with built-in serialization capabilities.
The C-Two framework automatically handles serialization/deserialization when this type
is detected as a parameter or return type in CRM methods (e.g., the return type of
Grid.get_grid_infos()).
"""
def serialize(data: list[GridAttribute]) -> bytes:
schema = pa.schema([
pa.field('attribute_bytes', pa.list_(pa.binary())),
])
data_dict = {
'attribute_bytes': [GridAttribute.serialize(grid) for grid in data]
}
table = pa.Table.from_pylist([data_dict], schema=schema)
return serialize_from_table(table)
def deserialize(arrow_bytes: bytes) -> list[GridAttribute]:
table = deserialize_to_table(arrow_bytes)
grid_bytes = table.column('attribute_bytes').to_pylist()[0]
return [GridAttribute.deserialize(grid_byte) for grid_byte in grid_bytes]
# Helpers ##################################################
def serialize_from_table(table: pa.Table) -> bytes:
sink = pa.BufferOutputStream()
with pa.ipc.new_stream(sink, table.schema) as writer:
writer.write_table(table)
binary_data = sink.getvalue().to_pybytes()
return binary_data
def deserialize_to_rows(serialized_data: bytes) -> dict:
buffer = pa.py_buffer(serialized_data)
with pa.ipc.open_stream(buffer) as reader:
table = reader.read_all()
return table.to_pylist()Deploy the CRM as a networked service with automatic protocol detection:
# Resource initialization and server configuration
grid = Grid(epsg=2326, bounds=[...], first_size=[64.0, 64.0], subdivide_rules=[...])
# TCP server (network-based, cross-machine)
server = cc.rpc.Server("tcp://localhost:5555", grid)
# HTTP server (web-compatible, cross-platform)
server = cc.rpc.Server("http://localhost:8000", grid)
# Memory server (shared memory, high-performance local communication)
server = cc.rpc.Server("memory://grid_region", grid)
# Same server lifecycle management regardless of protocol
server.start()
print('CRM server is running. Press Ctrl+C to stop.')
try:
# Wait for server termination with a timeout or not
server.wait_for_termination(timeout=10)
print('Timeout reached, stopping CRM server...')
except KeyboardInterrupt:
print('Keyboard interrupt received, stopping CRM server...')
finally:
server.stop()
print('CRM server stopped.')The framework provides seamless protocol-transparent connectivity to CRM services.
# TCP connection (network-based)
with cc.compo.runtime.connect_crm('tcp://localhost:5555', IGrid) as grid:
infos = grid.get_grid_infos(1, [0, 1, 2])
keys = grid.subdivide_grids([1, 1], [0, 1])
print(f'Retrieved {len(infos)} grid attributes, generated {len(keys)} subdivisions')
# HTTP connection (web-compatible)
with cc.compo.runtime.connect_crm('http://localhost:8000', IGrid) as grid:
infos = grid.get_grid_infos(1, [0, 1, 2])
keys = grid.subdivide_grids([1, 1], [0, 1])
print(f'Retrieved {len(infos)} grid attributes, generated {len(keys)} subdivisions')
# Memory connection (shared memory, high-performance local)
with cc.compo.runtime.connect_crm('memory://grid_region', IGrid) as grid:
infos = grid.get_grid_infos(1, [0, 1, 2])
keys = grid.subdivide_grids([1, 1], [0, 1])
print(f'Retrieved {len(infos)} grid attributes, generated {len(keys)} subdivisions')@cc.compo.runtime.connect
def process_grids(grid: IGrid, target_level: int) -> list[str]:
"""Reusable component for grid processing operations"""
infos = grid.get_grid_infos(target_level, [0, 1, 2, 3, 4])
candidates = [info.global_id for info in infos if hasattr(info, 'elevation') and info.elevation > 0]
return grid.subdivide_grids([target_level] * len(candidates), candidates)
# Protocol is determined automatically by the address
result = process_grids(1, crm_address='tcp://localhost:5555') # Uses TCP
result = process_grids(1, crm_address='http://localhost:8000') # Uses HTTP
result = process_grids(1, crm_address='memory://grid_region') # Uses Memory
# Or using a context manager with automatic protocol detection
with cc.compo.runtime.connect_crm('memory://grid_region'):
result = process_grids(1)Integrate with external systems using the Model Context Protocol (MCP):
from mcp.server.fastmcp import FastMCP
import compo # Component module
mcp = FastMCP('GridAgent', instructions=cc.mcp.CC_INSTRUCTION)
cc.mcp.register_mcp_tools_from_compo_module(mcp, compo)
if __name__ == '__main__':
mcp.run()C-Two is designed for resource-oriented distributed computing, particularly suitable for:
- Geographic Information Systems: Spatial data processing, map tile generation, geographic analysis with distributed resource access
- Scientific Computing: Computational models requiring persistent state and distributed access to computational resources
- Simulation Systems: Multi-agent simulations with distributed resource management and state synchronization
- Data Processing Pipelines: Stateful data transformation services with distributed coordination
- Computational Resource Management: Systems requiring dynamic allocation and distributed access to computational resources
- Microservices with State: Service architectures where services maintain persistent computational state
@icrm: Interface definition for remote resource specifications@iicrm: Implementation marker for Core Resource Models@transferable: Custom serializable data type definition@auto_transfer: Automatic serialization based on type annotations@connect: Connection injection for component functions
connect_crm: Context manager for CRM connection lifecycle management with automatic protocol detectionServer: CRM service deployment infrastructure with multi-protocol supportClient: Remote resource access client with transparent protocol handling
- MCP Tools: Model Context Protocol integration for external system communication
- Python ≥ 3.10
- Type annotation support
- Network connectivity for distributed deployment (TCP, HTTP protocols)
C-Two provides a principled approach to distributed resource computing through type-safe abstractions and protocol-agnostic communication, enabling developers to build robust distributed systems with persistent computational resources.