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

ReefLifecycleError

Raised when an active Reef operation is interrupted by its lifecycle.

SporeValidationError

Raised when spore validation fails.
exception praval.core.reef.SporeValidationError[source]

Bases: Exception

Raised when spore validation fails.

exception praval.core.reef.ReefLifecycleError[source]

Bases: RuntimeError

Raised when an active Reef operation is interrupted by its lifecycle.

class praval.core.reef.SporeType(*values)[source]

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, trace_context=None, run_id=None, idempotency_key=None)[source]

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)

  • trace_context (Dict[str, 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
trace_context: Dict[str, str] | None = None
run_id: str | None = None
idempotency_key: str | None = None
validate(max_size=None)[source]

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()[source]

Get the size of the knowledge payload in bytes.

Return type:

int

to_json()[source]

Serialize spore to JSON for transmission.

Return type:

str

classmethod from_json(json_str)[source]

Deserialize spore from JSON.

Return type:

Spore

Parameters:

json_str (str)

is_expired()[source]

Check if spore has expired.

Return type:

bool

add_knowledge_reference(reference_id)[source]

Return a new spore with an added knowledge reference.

Return type:

Spore

Parameters:

reference_id (str)

add_data_reference(reference_uri)[source]

Return a new spore with an added data reference.

Return type:

Spore

Parameters:

reference_uri (str)

has_knowledge_references()[source]

Check if spore has knowledge references

Return type:

bool

has_data_references()[source]

Check if spore has data references

Return type:

bool

has_any_references()[source]

Check if spore has any kind of references

Return type:

bool

get_spore_size_estimate()[source]

Estimate spore size for lightweight transmission

Return type:

int

to_amqp_message()[source]

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

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, trace_context=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)

  • trace_context (Dict[str, str] | None)

  • run_id (str | None)

  • idempotency_key (str | None)

Return type:

None

class praval.core.reef.SubscriptionManager[source]

Bases: object

Manage subscriber handlers for a channel.

__init__()[source]
set_handler(agent_name, handler)[source]
Return type:

None

Parameters:

  • agent_name (str)

  • handler (Callable)

add_handler(agent_name, handler)[source]
Return type:

None

Parameters:

  • agent_name (str)

  • handler (Callable)

remove_agent(agent_name)[source]
Return type:

None

Parameters:

agent_name (str)

remove_handler(agent_name, handler)[source]

Remove one handler without disturbing the agent’s other subscribers.

Return type:

None

Parameters:

  • agent_name (str)

  • handler (Callable)

get_handlers(agent_name)[source]
Return type:

List[Callable]

Parameters:

agent_name (str)

iter_broadcast(exclude_agent=None)[source]
Parameters:

exclude_agent (str | None)

counts()[source]
Return type:

Dict[str, int]

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

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)[source]
Parameters:

  • name (str)

  • max_capacity (int)

  • max_workers (int)

  • executor (ThreadPoolExecutor | None)

  • batch_size (int)

spores: deque
send_spore(spore)[source]

Send a spore through this channel.

Return type:

bool

Parameters:

spore (Spore)

subscribe(agent_name, handler, replace=True)[source]

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, handler=None)[source]

Unsubscribe an agent or one specific handler from this channel.

Return type:

None

Parameters:

  • agent_name (str)

  • handler (Callable[[Spore], None] | None)

get_spores_for_agent(agent_name, limit=10)[source]

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

Return type:

List[Spore]

Parameters:

  • agent_name (str)

  • limit (int)

cleanup_expired()[source]

Remove expired spores from the channel.

Return type:

int

get_stats()[source]

Get channel statistics.

Return type:

Dict[str, Any]

wait_for_completion(timeout=None)[source]

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

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

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

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

Create a new reef channel.

Return type:

ReefChannel

Parameters:

  • name (str)

  • max_capacity (int)

  • max_workers (int | None)

  • batch_size (int)

get_channel(name)[source]

Get a reef channel by name.

Return type:

Optional[ReefChannel]

Parameters:

name (str)

remove_channel_if_unused(name)[source]

Remove a channel that has no subscribers and shut it down.

The default channel is never removed. The subscriber check and the removal happen under the reef lock, which subscribe() also holds while registering a local handler, so a concurrent subscription is never left on a removed channel.

Parameters:

name (str) – Channel name

Return type:

bool

Returns:

True if the channel was removed, False if it does not exist, is the default channel, or still has subscribers.

async initialize_backend(config=None)[source]

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()[source]

Shutdown the backend (async operation for distributed backends).

Return type:

None

send_spore(spore, channel=None)[source]

Route an existing spore through authorization and the active backend.

Return type:

str

Parameters:

  • spore (Spore)

  • channel (str)

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, *, metadata=None, payload=None, content_parts=None, data_references=None, schema_version='1.0', correlation_id=None, causation_id=None, trace_id=None, trace_context=None, run_id=None, idempotency_key=None, expires_at=None)[source]

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)

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

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

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

  • data_references (List[str] | None)

  • schema_version (str)

  • correlation_id (str | None)

  • causation_id (str | None)

  • trace_id (str | None)

  • trace_context (Dict[str, str] | None)

  • run_id (str | None)

  • idempotency_key (str | None)

  • expires_at (datetime | None)

broadcast(from_agent, knowledge, channel=None)[source]

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

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, *, priority=5, metadata=None, payload=None, content_parts=None, knowledge_references=None, data_references=None, schema_version='1.0', correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None, expires_at=None)[source]

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)

  • priority (int)

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

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

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

  • knowledge_references (List[str] | None)

  • data_references (List[str] | None)

  • schema_version (str)

  • correlation_id (str | None)

  • causation_id (str | None)

  • trace_id (str | None)

  • run_id (str | None)

  • idempotency_key (str | None)

  • expires_at (datetime | None)

reply(from_agent, to_agent, response, reply_to_spore_id, channel=None, *, priority=5, expires_in_seconds=None, expires_at=None, metadata=None, payload=None, content_parts=None, knowledge_references=None, data_references=None, schema_version='1.0', correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None)[source]

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)

  • priority (int)

  • expires_in_seconds (float | None)

  • expires_at (datetime | None)

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

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

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

  • knowledge_references (List[str] | None)

  • data_references (List[str] | None)

  • schema_version (str)

  • correlation_id (str | None)

  • causation_id (str | None)

  • trace_id (str | None)

  • run_id (str | None)

  • idempotency_key (str | None)

request_and_wait(from_agent, to_agent, request, channel=None, expires_in_seconds=300, timeout=30.0, *, on_notification=None, priority=5, metadata=None, payload=None, content_parts=None, knowledge_references=None, data_references=None, schema_version='1.0', correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None, expires_at=None)[source]

Send a request and block for its strictly matched final response.

Return type:

Spore

Parameters:

  • from_agent (str)

  • to_agent (str)

  • request (Dict[str, Any])

  • channel (str)

  • expires_in_seconds (int)

  • timeout (float | None)

  • on_notification (Callable[[Spore], Any] | None)

  • priority (int)

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

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

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

  • knowledge_references (List[str] | None)

  • data_references (List[str] | None)

  • schema_version (str)

  • correlation_id (str | None)

  • causation_id (str | None)

  • trace_id (str | None)

  • run_id (str | None)

  • idempotency_key (str | None)

  • expires_at (datetime | None)

async arequest_and_wait(from_agent, to_agent, request, channel=None, expires_in_seconds=300, timeout=30.0, *, on_notification=None, priority=5, metadata=None, payload=None, content_parts=None, knowledge_references=None, data_references=None, schema_version='1.0', correlation_id=None, causation_id=None, trace_id=None, run_id=None, idempotency_key=None, expires_at=None)[source]

Asynchronously send a request and await its matched final response.

Return type:

Spore

Parameters:

  • from_agent (str)

  • to_agent (str)

  • request (Dict[str, Any])

  • channel (str)

  • expires_in_seconds (int)

  • timeout (float | None)

  • on_notification (Callable[[Spore], Any] | None)

  • priority (int)

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

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

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

  • knowledge_references (List[str] | None)

  • data_references (List[str] | None)

  • schema_version (str)

  • correlation_id (str | None)

  • causation_id (str | None)

  • trace_id (str | None)

  • run_id (str | None)

  • idempotency_key (str | None)

  • expires_at (datetime | None)

reply_to_request(request_spore, response, **options)[source]

Reply with lifecycle fields derived from the originating request.

Return type:

str

Parameters:

  • request_spore (Spore)

  • response (Dict[str, Any])

  • options (Any)

notify_request(request_spore, notification, **options)[source]

Send a correlated progress notification for an active request.

Return type:

str

Parameters:

  • request_spore (Spore)

  • notification (Dict[str, Any])

  • options (Any)

forward_request(request_spore, to_agent, request=None, *, channel=None, priority=None, expires_in_seconds=None, expires_at=None, metadata=None, payload=None, content_parts=None, knowledge_references=None, data_references=None, schema_version=None)[source]

Forward a request without extending its lifecycle.

Return type:

str

Parameters:

  • request_spore (Spore)

  • to_agent (str)

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

  • channel (str)

  • priority (int | None)

  • expires_in_seconds (float | None)

  • expires_at (datetime | None)

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

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

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

  • knowledge_references (List[str] | None)

  • data_references (List[str] | None)

  • schema_version (str | None)

subscribe(agent_name, handler, channel=None, replace=True)[source]

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()[source]

Get statistics about the reef network.

Return type:

Dict[str, Any]

wait_for_completion(timeout=None)[source]

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

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

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

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

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)

praval.core.reef.get_reef()[source]

Get the global reef instance.

Return type:

Reef

praval.core.reef.reset_reef()[source]

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