Cortex Mesh SDK

The Cortex Mesh SDK is a portable, transport-agnostic framework for building distributed agentic networks. It provides the core orchestration logic for the Cortex Hub and its remote Agent Nodes, ensuring reliable communication, health monitoring, and task dispatch regardless of the underlying protocol.

Architecture

The SDK is built on a layered architecture that decouples application logic from the transport layer:

1. Core Engines (mesh_core):
   • MeshServerCore: Manages node registrations, heartbeats, and message routing on the Hub side.
   • MeshNodeCore: Manages the node lifecycle, reconnection logic, and task processing on the Agent side.
2. Transport Layer (IMeshTransport):
   • A standardized interface for sending and receiving messages.
   • GrpcMeshTransport: Production-grade implementation using gRPC bidirectional streams.
   • MockTransport: Used for unit testing and local development without network overhead.
3. Protobuf Definitions (protos):
   • Centralized definitions for all mesh communication (Tasks, File Sync, Health, etc.).

Key Features

• Transport Agnostic: Easily swap gRPC for WebSockets, MQTT, or Mock transports.
• Auto-Reconnection: Built-in exponential backoff and state recovery for unstable network conditions.
• Multiplexed Priority Queues: Built-in support for prioritizing Admin/Control tasks over background File Sync.
• Health & Telemetry: Dedicated health streams for high-frequency metric reporting with minimal overhead.

Usage

For Hub Developers (Server)

from mesh_core.server_engine import MeshServerCore
from mesh_core.transport_grpc import GrpcServerTransport

mesh = MeshServerCore()

# Bind events
mesh.on_node_online = lambda node: print(f"Node {node.node_id} is online")
mesh.on_message_received = lambda node_id, msg: handle_msg(node_id, msg)

# In your gRPC TaskStream handler:
transport = GrpcServerTransport(context, signer=my_signer)
mesh.register_node(node_id, user_id, metadata, transport)

For Agent Developers (Node)

from mesh_core.node_engine import MeshNodeCore
from mesh_core.transport_grpc import GrpcMeshTransport

transport = GrpcMeshTransport(node_id, stub_factory, auth_token)
node = MeshNodeCore(node_id, transport)

# Bind task handlers
node.on_task = lambda task: execute_task(task)

# Start the engine
node.start()

Testing

The SDK includes a comprehensive test suite using the MockTransport to validate orchestration logic without network dependencies.

# Run component tests
PYTHONPATH=. pytest tests/