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])	The Reef manages all communication channels and facilitates agent communication.
ReefChannel(name[, max_capacity, max_workers])	A message channel within the reef.
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.

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)

Bases: object

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

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])

• created_at (datetime)

• expires_at (datetime | None)

• priority (int)

• reply_to (str | None)

• metadata (Dict[str, Any])

• knowledge_references (List[str])

• data_references (List[str])

id: str

spore_type: SporeType

from_agent: str

to_agent: Optional[str]

knowledge: Dict[str, Any]

created_at: datetime

expires_at: Optional[datetime] = None

priority: int = 5

reply_to: Optional[str] = None

metadata: Dict[str, Any] = None

knowledge_references: List[str] = None

data_references: List[str] = None

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)

Add a reference to stored knowledge

Parameters:

reference_id (str)

add_data_reference(reference_uri)

Add a reference to storage system data

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:

aio_pika.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 (aio_pika.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)

Parameters:

• id (str)

• spore_type (SporeType)

• from_agent (str)

• to_agent (str | None)

• knowledge (Dict[str, Any])

• created_at (datetime)

• expires_at (datetime | None)

• priority (int)

• reply_to (str | None)

• metadata (Dict[str, Any])

• knowledge_references (List[str])

• data_references (List[str])

Return type:

None

class praval.core.reef.ReefChannel(name, max_capacity=1000, max_workers=4)

Bases: object

A message channel within the reef.

Like channels in a coral reef, they: - Have directional flow patterns - Can carry multiple spores simultaneously - Have capacity limits (to prevent overwhelming) - Can experience turbulence (message loss/delays)

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int)

__init__(name, max_capacity=1000, max_workers=4)

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int)

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)

Shutdown the channel’s thread pool.

Return type:

None

Parameters:

wait (bool)

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

Bases: object

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)

__init__(default_max_workers=4, backend=None)

Initialize Reef with optional backend.

Parameters:

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

• backend – ReefBackend instance (defaults to InMemoryBackend)

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

Create a new reef channel.

Return type:

ReefChannel

Parameters:

• name (str)

• max_capacity (int)

• max_workers (int | None)

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)

Shutdown the reef and all its channels.

Return type:

None

Parameters:

wait (bool)

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

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