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.

Classes

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

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

Subscribe an agent to receive spores from this channel.

Return type:

None

Parameters:

• agent_name (str)

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

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]

shutdown(wait=True)

Shutdown the channel’s thread pool.

Return type:

None

Parameters:

wait (bool)

class praval.core.reef.Reef(default_max_workers=4)

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

Parameters:

default_max_workers (int)

__init__(default_max_workers=4)

Parameters:

default_max_workers (int)

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)

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)

Subscribe an agent to receive spores from a channel.

Return type:

None

Parameters:

• agent_name (str)

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

• channel (str)

get_network_stats()

Get statistics about the reef network.

Return type:

Dict[str, Any]

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