praval.core.reef

Reef Communication System for Praval Framework.

Like coral reefs facilitate communication between polyps through chemical and biological signals, this system enables knowledge-first communication between agents through structured JSON message queues.

Components: - Spores: JSON messages containing knowledge, data, or requests - ReefChannel: Named message channels within the reef - Reef: The message queue network connecting all agents

Functions

get_reef()	Get the global reef instance.
reset_reef()	Reset the global reef instance to a clean state.

Classes

Reef([default_max_workers, backend, ...])	Facade for backwards-compatible Reef API.
ReefChannel(name[, max_capacity, ...])
ReefCore([default_max_workers, backend, ...])	Core Reef implementation (transport and channel management).
Spore(id, spore_type, from_agent, to_agent, ...)	A spore is a knowledge-carrying message that flows through the reef.
SporeType(*values)	Types of spores that can flow through the reef.
SubscriptionManager()	Manage subscriber handlers for a channel.

Exceptions

SporeValidationError

Raised when spore validation fails.

exception praval.core.reef.SporeValidationError

Bases: Exception

Raised when spore validation fails.

class praval.core.reef.SporeType(*values)

Bases: Enum

Types of spores that can flow through the reef.

KNOWLEDGE = 'knowledge'

REQUEST = 'request'

RESPONSE = 'response'

BROADCAST = 'broadcast'

NOTIFICATION = 'notification'

class praval.core.reef.Spore(id, spore_type, from_agent, to_agent, knowledge, created_at, expires_at=None, priority=5, reply_to=None, metadata=None, knowledge_references=None, data_references=None, schema_version='1.0', payload=None, content_parts=None, correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None)

Bases: object

A spore is a knowledge-carrying message that flows through the reef.

Spores are immutable. To add references, use add_knowledge_reference()/add_data_reference(), which return new Spore instances.

Like biological spores, each carries: - Genetic material (knowledge/data) - Identification markers (metadata) - Survival instructions (processing hints)

Spores can carry either direct knowledge or lightweight references to knowledge stored in vector memory, following the principle that “light spores travel far.”

Parameters:

• id (str)

• spore_type (SporeType)

• from_agent (str)

• to_agent (str | None)

• knowledge (Dict[str, Any] | None)

• created_at (datetime)

• expires_at (datetime | None)

• priority (int)

• reply_to (str | None)

• metadata (Dict[str, Any] | None)

• knowledge_references (List[str])

• data_references (List[str])

• schema_version (str)

• payload (Dict[str, Any] | None)

• content_parts (List[Dict[str, Any]])

• correlation_id (str | None)

• causation_id (str | None)

• trace_id (str | None)

• run_id (str | None)

• idempotency_key (str | None)

id: str

spore_type: SporeType

from_agent: str

to_agent: str | None

knowledge: Dict[str, Any] | None

created_at: datetime

expires_at: datetime | None = None

priority: int = 5

reply_to: str | None = None

metadata: Dict[str, Any] | None = None

knowledge_references: List[str] = None

data_references: List[str] = None

schema_version: str = '1.0'

payload: Dict[str, Any] | None = None

content_parts: List[Dict[str, Any]] = None

correlation_id: str | None = None

causation_id: str | None = None

trace_id: str | None = None

run_id: str | None = None

idempotency_key: str | None = None

validate(max_size=None)

Validate the spore payload.

Parameters:

• max_size (int) – Maximum allowed payload size in bytes (default:

• MAX_SPORE_SIZE_BYTES)

Raises:

SporeValidationError – If validation fails

Return type:

None

get_payload_size()

Get the size of the knowledge payload in bytes.

Return type:

int

to_json()

Serialize spore to JSON for transmission.

Return type:

str

classmethod from_json(json_str)

Deserialize spore from JSON.

Return type:

Spore

Parameters:

json_str (str)

is_expired()

Check if spore has expired.

Return type:

bool

add_knowledge_reference(reference_id)

Return a new spore with an added knowledge reference.

Return type:

Spore

Parameters:

reference_id (str)

add_data_reference(reference_uri)

Return a new spore with an added data reference.

Return type:

Spore

Parameters:

reference_uri (str)

has_knowledge_references()

Check if spore has knowledge references

Return type:

bool

has_data_references()

Check if spore has data references

Return type:

bool

has_any_references()

Check if spore has any kind of references

Return type:

bool

get_spore_size_estimate()

Estimate spore size for lightweight transmission

Return type:

int

to_amqp_message()

Convert Spore to AMQP message with metadata in headers.

Design: - Body: Knowledge payload (JSON serialized) - Headers: Spore metadata (spore_id, type, from_agent, etc.) - Properties: AMQP message properties (priority, TTL, etc.)

This makes Spore the native AMQP format, eliminating intermediate conversions.

Returns:

AMQP message ready for publication

Return type:

Message

Raises:

ImportError – If aio-pika is not installed

classmethod from_amqp_message(amqp_msg)

Create Spore directly from AMQP message.

Reconstructs a Spore object from AMQP message headers and body, with zero intermediate conversions (AMQP message directly becomes Spore).

Parameters:

amqp_msg (Message) – aio_pika.Message from AMQP broker

Returns:

Reconstructed spore object with all metadata

Return type:

Spore

Raises:

• ImportError – If aio-pika is not installed

• ValueError – If required spore headers are missing

__init__(id, spore_type, from_agent, to_agent, knowledge, created_at, expires_at=None, priority=5, reply_to=None, metadata=None, knowledge_references=None, data_references=None, schema_version='1.0', payload=None, content_parts=None, correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None)

Parameters:

• id (str)

• spore_type (SporeType)

• from_agent (str)

• to_agent (str | None)

• knowledge (Dict[str, Any] | None)

• created_at (datetime)

• expires_at (datetime | None)

• priority (int)

• reply_to (str | None)

• metadata (Dict[str, Any] | None)

• knowledge_references (List[str])

• data_references (List[str])

• schema_version (str)

• payload (Dict[str, Any] | None)

• content_parts (List[Dict[str, Any]])

• correlation_id (str | None)

• causation_id (str | None)

• trace_id (str | None)

• run_id (str | None)

• idempotency_key (str | None)

Return type:

None

class praval.core.reef.SubscriptionManager

Bases: object

Manage subscriber handlers for a channel.

__init__()

set_handler(agent_name, handler)

Return type:

None

Parameters:

• agent_name (str)

• handler (Callable)

add_handler(agent_name, handler)

Return type:

None

Parameters:

• agent_name (str)

• handler (Callable)

remove_agent(agent_name)

Return type:

None

Parameters:

agent_name (str)

get_handlers(agent_name)

Return type:

List[Callable]

Parameters:

agent_name (str)

iter_broadcast(exclude_agent=None)

Parameters:

exclude_agent (str | None)

counts()

Return type:

Dict[str, int]

class praval.core.reef.ReefChannel(name, max_capacity=1000, max_workers=4, executor=None, batch_size=1)

Bases: object

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int)

• executor (ThreadPoolExecutor | None)

• batch_size (int)

__init__(name, max_capacity=1000, max_workers=4, executor=None, batch_size=1)

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int)

• executor (ThreadPoolExecutor | None)

• batch_size (int)

spores: deque

send_spore(spore)

Send a spore through this channel.

Return type:

bool

Parameters:

spore (Spore)

subscribe(agent_name, handler, replace=True)

Subscribe an agent to receive spores from this channel.

Parameters:

• agent_name (str) – Name of the agent subscribing

• handler (Callable[[Spore], None]) – Callback function to handle received spores

• replace (bool) – If True (default), replaces existing handlers for this agent. If False, adds handler to the list (useful for multiple handlers).

Return type:

None

Note

Default behavior (replace=True) ensures that re-registering an agent in interactive environments (like Jupyter notebooks) doesn’t create duplicate subscriptions. Set replace=False if you intentionally want multiple handlers for the same agent.

unsubscribe(agent_name)

Unsubscribe an agent from this channel.

Return type:

None

Parameters:

agent_name (str)

get_spores_for_agent(agent_name, limit=10)

Get recent spores for a specific agent (polling interface).

Return type:

List[Spore]

Parameters:

• agent_name (str)

• limit (int)

cleanup_expired()

Remove expired spores from the channel.

Return type:

int

get_stats()

Get channel statistics.

Return type:

Dict[str, Any]

wait_for_completion(timeout=None)

Wait for all active handler executions to complete.

This method blocks until all currently running and pending handlers finish, including handlers that spawn new handlers (cascading messages).

Parameters:

timeout (Optional[float]) – Maximum time to wait in seconds. None means wait indefinitely.

Return type:

bool

Returns:

True if all handlers completed, False if timeout occurred.

shutdown(wait=True, timeout=30.0)

Shutdown the channel’s thread pool.

Parameters:

• wait (bool) – Whether to wait for pending handlers to complete

• timeout (float) – Maximum seconds to wait (only if wait=True)

Return type:

bool

Returns:

True if shutdown completed cleanly, False if timeout occurred

class praval.core.reef.ReefCore(default_max_workers=4, backend=None, use_shared_pool=True, auth_provider=None)

Bases: object

Core Reef implementation (transport and channel management).

The Reef manages all communication channels and facilitates agent communication.

Like a coral reef ecosystem, it: - Maintains multiple communication channels - Enables knowledge flow between polyps (agents) - Supports both direct and broadcast communication - Provides network health monitoring

The Reef uses pluggable backends for transport: - InMemoryBackend: Local agent communication (default) - RabbitMQBackend: Distributed agent communication - Future: HTTP, gRPC, Kafka, etc.

Agents work unchanged regardless of backend choice.

Message Routing: - When using InMemoryBackend: Messages routed through local ReefChannel - When using RabbitMQBackend (or other distributed): Messages routed through backend

Parameters:

• default_max_workers (int)

• use_shared_pool (bool)

• auth_provider (Callable[[str, Dict[str, Any]], bool] | None)

__init__(default_max_workers=4, backend=None, use_shared_pool=True, auth_provider=None)

Initialize Reef with optional backend.

Parameters:

• default_max_workers (int) – Thread workers per channel (InMemory only)

• backend – ReefBackend instance (defaults to InMemoryBackend)

• use_shared_pool (bool) – Share a single thread pool across channels

• auth_provider (Optional[Callable[[str, Dict[str, Any]], bool]]) – Optional authorization callback (action, context) -> bool

channels: Dict[str, ReefChannel]

create_channel(name, max_capacity=1000, max_workers=None, batch_size=1)

Create a new reef channel.

Return type:

ReefChannel

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int | None)

• batch_size (int)

get_channel(name)

Get a reef channel by name.

Return type:

Optional[ReefChannel]

Parameters:

name (str)

async initialize_backend(config=None)

Initialize the Reef backend (async operation for distributed backends).

Call this method to set up distributed backends like RabbitMQ. InMemoryBackend initializes immediately, so this is optional for local usage.

Parameters:

config (Optional[Dict[str, Any]]) – Backend-specific configuration (passed to backend.initialize())

Return type:

None

async close_backend()

Shutdown the backend (async operation for distributed backends).

Return type:

None

send(from_agent, to_agent, knowledge, spore_type=SporeType.KNOWLEDGE, channel=None, priority=5, expires_in_seconds=None, reply_to=None, knowledge_references=None, auto_reference_large_knowledge=True)

Send a spore through the reef.

Return type:

str

Parameters:

• from_agent (str)

• to_agent (str | None)

• knowledge (Dict[str, Any])

• spore_type (SporeType)

• channel (str)

• priority (int)

• expires_in_seconds (int | None)

• reply_to (str | None)

• knowledge_references (List[str] | None)

• auto_reference_large_knowledge (bool)

broadcast(from_agent, knowledge, channel=None)

Broadcast knowledge to all agents in the reef.

Return type:

str

Parameters:

• from_agent (str)

• knowledge (Dict[str, Any])

• channel (str)

system_broadcast(knowledge, channel=None)

Broadcast system-level messages to all agents in a channel.

Return type:

str

Parameters:

• knowledge (Dict[str, Any])

• channel (str)

request(from_agent, to_agent, request, channel=None, expires_in_seconds=300)

Send a knowledge request to another agent.

Return type:

str

Parameters:

• from_agent (str)

• to_agent (str)

• request (Dict[str, Any])

• channel (str)

• expires_in_seconds (int)

reply(from_agent, to_agent, response, reply_to_spore_id, channel=None)

Reply to a knowledge request.

Return type:

str

Parameters:

• from_agent (str)

• to_agent (str)

• response (Dict[str, Any])

• reply_to_spore_id (str)

• channel (str)

subscribe(agent_name, handler, channel=None, replace=True)

Subscribe an agent to receive spores from a channel.

Parameters:

• agent_name (str) – Name of the agent subscribing

• handler (Callable[[Spore], None]) – Callback function to handle received spores

• channel (str) – Channel name (uses default if None)

• replace (bool) – If True (default), replaces existing handlers for this agent. If False, adds handler to the list.

Return type:

None

Note

For distributed backends (RabbitMQ, etc.), this also subscribes to the backend’s message broker, enabling cross-process communication.

get_network_stats()

Get statistics about the reef network.

Return type:

Dict[str, Any]

wait_for_completion(timeout=None)

Wait for all active agent handlers to complete across all channels.

This method blocks until all currently running handlers finish, including cascading handlers triggered by broadcast() calls within agents.

Parameters:

timeout (Optional[float]) – Maximum time to wait in seconds. None means wait indefinitely.

Return type:

bool

Returns:

True if all handlers completed, False if timeout occurred.

Example

start_agents(researcher, summarizer, initial_data={…}) get_reef().wait_for_completion() # Block until all agents done get_reef().shutdown()

shutdown(wait=True, timeout=30.0)

Shutdown the reef and all its channels.

Parameters:

• wait (bool) – Whether to wait for pending handlers

• timeout (float) – Maximum total seconds to wait across all channels

Return type:

bool

Returns:

True if all channels shut down cleanly, False if timeout occurred

create_knowledge_reference_spore(from_agent, to_agent, knowledge_summary, knowledge_references, spore_type=SporeType.KNOWLEDGE, channel=None)

Create a lightweight spore with knowledge references

This follows the reef principle: “light spores travel far”

Return type:

str

Parameters:

• from_agent (str)

• to_agent (str | None)

• knowledge_summary (str)

• knowledge_references (List[str])

• spore_type (SporeType)

• channel (str)

resolve_knowledge_references(spore, memory_manager)

Resolve knowledge references in a spore to actual knowledge

Parameters:

• spore (Spore) – The spore with knowledge references

• memory_manager – Agent’s memory manager to resolve references

Return type:

Dict[str, Any]

Returns:

Combined knowledge from references

class praval.core.reef.Reef(default_max_workers=4, backend=None, use_shared_pool=True, auth_provider=None)

Bases: ReefCore

Facade for backwards-compatible Reef API.

Parameters:

• default_max_workers (int)

• use_shared_pool (bool)

• auth_provider (Callable[[str, Dict[str, Any]], bool] | None)

broadcast(from_agent, knowledge, **kwargs)

Broadcast knowledge to all agents in the reef.

send(from_agent, to_agent, knowledge, **kwargs)

Send a spore through the reef.

praval.core.reef.get_reef()

Get the global reef instance.

Return type:

Reef

praval.core.reef.reset_reef()

Reset the global reef instance to a clean state.

This is primarily used for testing to ensure test isolation. Clears all channels and reinitializes with just the default channel.

Return type:

None