API Reference 

Returns:

Agent’s response as a string

Raises:

ValueError – If message is empty or None
PravalError – If response generation fails

tool(func)[source]

Decorator to register a function as a tool for the agent.

Parameters:: func (Callable) – Function to register as a tool
Return type:: Callable
Returns:: The original function (unchanged)
Raises:: ValueError – If function lacks proper type hints

send_knowledge(to_agent, knowledge, channel='main')[source]

Send knowledge to another agent through the reef.

Parameters:

to_agent (str) – Name of the target agent
knowledge (Dict[str, Any]) – Knowledge data to send
channel (str) – Reef channel to use (default: “main”)

Return type:

Returns:

Spore ID of the sent message

broadcast_knowledge(knowledge, channel='main')[source]

Broadcast knowledge to all agents in the reef.

Parameters:

knowledge (Dict[str, Any]) – Knowledge data to broadcast
channel (str) – Reef channel to use (default: “main”)

Return type:

Returns:

Spore ID of the broadcast message

request_knowledge(from_agent, request, timeout=30)[source]

Request knowledge from another agent with timeout.

Parameters:

from_agent (str) – Name of the agent to request from
request (Dict[str, Any]) – Request data
timeout (int) – Timeout in seconds

Return type:

Optional[Dict[str, Any]]

Returns:

Response data or None if timeout

on_spore_received(spore)[source]

Handle received spores from the reef.

This is a default implementation that can be overridden by subclasses for custom spore handling.

Parameters:: spore – The received Spore object
Return type:: None

subscribe_to_channel(channel_name)[source]

Subscribe this agent to a reef channel.

Parameters:: channel_name (str) – Name of the channel to subscribe to
Return type:: None

unsubscribe_from_channel(channel_name)[source]

Unsubscribe this agent from a reef channel.

Parameters:: channel_name (str) – Name of the channel to unsubscribe from
Return type:: None

set_spore_handler(handler)[source]

Set a custom spore handler for this agent.

Parameters:: handler (Callable) – Function that takes a Spore object and handles it
Return type:: None

remember(content, importance=0.5, memory_type='short_term')[source]

Store a memory

Parameters:

content (str) – The content to remember
importance (float) – Importance score (0.0 to 1.0)
memory_type (str) – Type of memory (“short_term”, “semantic”, “episodic”)

Return type:

Optional[str]

Returns:

Memory ID if successful, None otherwise

recall(query, limit=5, similarity_threshold=0.1)[source]

Recall memories based on a query

Parameters:

query (str) – Search query
limit (int) – Maximum number of results
similarity_threshold (float) – Minimum similarity score

Return type:

List

Returns:

List of MemoryEntry objects

recall_by_id(memory_id)[source]

Recall a specific memory by ID (for resolving spore references)

Return type:: List
Parameters:: memory_id (str)

get_conversation_context(turns=10)[source]

Get recent conversation context

Return type:: List
Parameters:: turns (int)

create_knowledge_reference(content, importance=0.8)[source]

Create knowledge references for lightweight spores

Parameters:

content (str) – Knowledge content to store and reference
importance (float) – Importance threshold

Return type:

List[str]

Returns:

List of knowledge reference IDs

resolve_spore_knowledge(spore)[source]

Resolve knowledge references in a spore

Parameters:: spore – Spore object with potential knowledge references
Return type:: Dict[str, Any]
Returns:: Complete knowledge including resolved references

send_lightweight_knowledge(to_agent, large_content, summary, channel='main')[source]

Send large knowledge as lightweight spore with references

Parameters:

to_agent (str) – Target agent
large_content (str) – Large content to reference
summary (str) – Brief summary for the spore
channel (str) – Communication channel

Return type:

Returns:

Spore ID

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

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

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

Add a reference to stored knowledge

Parameters:: reference_id (str)

add_data_reference(reference_uri)[source]

Add a reference to storage system data

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

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

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

Parameters:

name (str)
max_capacity (int)
max_workers (int)

send_spore(spore)[source]

Send a spore through this channel.

Return type:: bool
Parameters:: spore (Spore)

subscribe(agent_name, handler)[source]

Subscribe an agent to receive spores from this channel.

Return type:

None

Parameters:

agent_name (str)
handler (Callable[[Spore], None])

unsubscribe(agent_name)[source]

Unsubscribe an agent from this channel.

Return type:: None
Parameters:: agent_name (str)

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]

shutdown(wait=True)[source]

Shutdown the channel’s thread pool.

Return type:: None
Parameters:: wait (bool)

class praval.core.reef.Reef(default_max_workers=4)[source]

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

Parameters:: default_max_workers (int)

create_channel(name, max_capacity=1000, max_workers=None)[source]

Create a new reef channel.

Return type:

ReefChannel

Parameters:

name (str)
max_capacity (int)
max_workers (int | None)

get_channel(name)[source]

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

Send a spore through the reef.

Return type:

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

Broadcast knowledge to all agents in the reef.

Return type:

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:

Parameters:

knowledge (Dict[str, Any])
channel (str)

request(from_agent, to_agent, request, channel=None, expires_in_seconds=300)[source]

Send a knowledge request to another agent.

Return type:

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

Reply to a knowledge request.

Return type:

Parameters:

from_agent (str)
to_agent (str)
response (Dict[str, Any])
reply_to_spore_id (str)
channel (str)

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

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

Get statistics about the reef network.

Return type:: Dict[str, Any]

shutdown(wait=True)[source]

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

Create a lightweight spore with knowledge references

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

Return type:

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

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

Get the global reef instance.

Return type:: Reef

Agent and Tool Registry for Praval Framework.

Provides a global registry for tracking agents and tools across the system, enabling better coordination and discovery in multi-agent applications.

class praval.core.registry.PravalRegistry[source]

Bases: object

Global registry for agents and tools in Praval applications.

__init__()[source]

register_agent(agent)[source]

Register an agent in the global registry.

Parameters:: agent (Agent) – Agent instance to register
Return type:: Agent
Returns:: The registered agent

get_agent(name)[source]

Get an agent by name from the registry.

Return type:: Optional[Agent]
Parameters:: name (str)

get_all_agents()[source]

Get all registered agents.

Return type:: Dict[str, Agent]

get_tool(tool_name)[source]

Get a tool by name from the registry.

Return type:: Optional[Dict[str, Any]]
Parameters:: tool_name (str)

get_tools_by_agent(agent_name)[source]

Get all tools for a specific agent.

Return type:: Dict[str, Dict[str, Any]]
Parameters:: agent_name (str)

list_agents()[source]

List names of all registered agents.

Return type:: List[str]

list_tools()[source]

List names of all registered tools.

Return type:: List[str]

clear()[source]: Clear all registered agents and tools.

praval.core.registry.register_agent(agent)[source]

Register an agent in the global registry.

Return type:: Agent
Parameters:: agent (Agent)

praval.core.registry.get_registry()[source]

Get the global registry instance.

Return type:: PravalRegistry

Custom exceptions for the Praval framework.

These exceptions provide clear error handling and debugging information for common issues in LLM agent operations.

exception praval.core.exceptions.PravalError[source]

Bases: Exception

Base exception for all Praval-related errors.

exception praval.core.exceptions.ProviderError[source]

Bases: PravalError

Raised when there are issues with LLM provider operations.

exception praval.core.exceptions.ConfigurationError[source]

Bases: PravalError

Raised when there are configuration validation issues.

exception praval.core.exceptions.ToolError[source]

Bases: PravalError

Raised when there are issues with tool registration or execution.

exception praval.core.exceptions.StateError[source]

Bases: PravalError

Raised when there are issues with state persistence operations.

Decorators Module

Decorator-based Agent API for Praval Framework.

This module provides a Pythonic decorator interface for creating agents that automatically handle reef communication and coordination.

Example

@agent(“explorer”, channel=”knowledge”) def explore_concepts(spore):

concepts = chat(“Find concepts related to: “ + spore.knowledge.get(“concept”, “”)) return {“discovered”: concepts.split(“,”)}

praval.decorators.agent(name=None, channel=None, system_message=None, auto_broadcast=True, responds_to=None, memory=False, knowledge_base=None)[source]

Decorator that turns a function into an autonomous agent.

Parameters:

name (Optional[str]) – Agent name (defaults to function name)
channel (Optional[str]) – Channel to subscribe to (defaults to name + “_channel”)
system_message (Optional[str]) – System message (defaults to function docstring)
auto_broadcast (bool) – Whether to auto-broadcast return values
responds_to (Optional[List[str]]) – List of message types this agent responds to (None = all messages)
memory (Union[bool, Dict[str, Any]]) – Memory configuration - True for defaults, dict for custom config, False to disable
knowledge_base (Optional[str]) – Path to knowledge base files for auto-indexing

Examples

Basic agent: @agent(“explorer”, channel=”knowledge”, responds_to=[“concept_request”]) def explore_concepts(spore):

‘’’Find related concepts and broadcast discoveries.’’’ concepts = chat(“Related to: “ + spore.knowledge.get(“concept”, “”)) return {“type”: “discovery”, “discovered”: concepts.split(“,”)}

Agent with memory: @agent(“researcher”, memory=True) def research_agent(spore):

‘’’Research agent with memory capabilities.’’’ query = spore.knowledge.get(“query”) # Remember the research research_agent.remember(f”Researched: {query}”) # Recall similar past research past_research = research_agent.recall(query) return {“research”: “completed”, “past_similar”: len(past_research)}

Agent with knowledge base: @agent(“expert”, memory=True, knowledge_base=”./knowledge/”) def expert_agent(spore):

‘’’Expert with pre-loaded knowledge base.’’’ question = spore.knowledge.get(“question”) relevant = expert_agent.recall(question, limit=3) return {“answer”: [r.content for r in relevant]}

praval.decorators.chat(message, timeout=10.0)[source]

Quick chat function that uses the current agent’s LLM with timeout support. Can only be used within @agent decorated functions.

Parameters:

message (str) – Message to send to the LLM
timeout (float) – Maximum time to wait for response in seconds

Return type:

Returns:

LLM response as string

Raises:

RuntimeError – If called outside of an @agent function
TimeoutError – If LLM call exceeds timeout

async praval.decorators.achat(message, timeout=10.0)[source]

Async version of chat function for use within async agent handlers.

Parameters:

message (str) – Message to send to the LLM
timeout (float) – Maximum time to wait for response in seconds

Return type:

Returns:

LLM response as string

Raises:

RuntimeError – If called outside of an @agent function
TimeoutError – If LLM call exceeds timeout

praval.decorators.broadcast(data, channel=None, message_type=None)[source]

Quick broadcast function that uses the current agent’s communication. Can only be used within @agent decorated functions.

Parameters:

data (Dict[str, Any]) – Data to broadcast
channel (Optional[str]) – Channel to broadcast to (defaults to agent’s channel)
message_type (Optional[str]) – Message type to set (automatically added to data)

Return type:

Returns:

Spore ID of the broadcast message

Raises:

RuntimeError – If called outside of an @agent function

praval.decorators.get_agent_info(agent_func)[source]

Get information about an @agent decorated function.

Parameters:: agent_func (Callable) – Function decorated with @agent
Return type:: Dict[str, Any]
Returns:: Dictionary with agent metadata

Composition utilities for decorator-based agents.

This module provides utilities for composing and orchestrating agents decorated with the @agent decorator.

praval.composition.agent_pipeline(*agents, channel='pipeline')[source]

Compose agents into a pipeline that processes data sequentially.

Parameters:

*agents (Callable) – Functions decorated with @agent
channel (str) – Channel name for pipeline communication

Return type:

Callable

Returns:

Function that triggers the pipeline with initial data

Example

pipeline = agent_pipeline(explorer, analyzer, reporter) pipeline({“task”: “analyze sentiment”})

praval.composition.conditional_agent(condition_func)[source]

Decorator for conditional agent execution.

Parameters:: condition_func (Callable[[Any], bool]) – Function that takes a spore and returns bool

Example

@conditional_agent(lambda spore: spore.knowledge.get(“priority”) == “high”) @agent(“urgent_processor”) def process_urgent(spore):

return {“processed”: True}

praval.composition.throttled_agent(delay_seconds)[source]

Decorator to throttle agent execution.

Parameters:: delay_seconds (float) – Minimum seconds between executions

Example

@throttled_agent(2.0) # Max once every 2 seconds @agent(“slow_processor”) def process_slowly(spore):

return {“processed”: True}

class praval.composition.AgentSession(session_name)[source]

Bases: object

Context manager for coordinated agent sessions.

Example

with AgentSession(“knowledge_mining”) as session:: session.add_agents(explorer, analyzer, curator) session.broadcast({“task”: “mine concepts about AI”})

Parameters:: session_name (str)

__init__(session_name)[source]

Parameters:: session_name (str)

add_agent(agent_func)[source]

Add an agent to this session.

Return type:: AgentSession
Parameters:: agent_func (Callable)

add_agents(*agent_funcs)[source]

Add multiple agents to this session.

Return type:: AgentSession
Parameters:: agent_funcs (Callable)

broadcast(data)[source]

Broadcast data to all agents in this session.

Return type:: str
Parameters:: data (Dict[str, Any])

get_stats()[source]

Get session statistics.

Return type:: Dict[str, Any]

praval.composition.start_agents(*agent_funcs, initial_data=None, channel='startup')[source]

Convenience function to start multiple agents with initial data.

Parameters:

*agent_funcs (Callable) – Functions decorated with @agent
initial_data (Optional[Dict[str, Any]]) – Initial data to broadcast (optional)
channel (str) – Channel to use for startup communication

Return type:

Returns:

Spore ID of startup broadcast

Example

start_agents(explorer, analyzer, curator,: initial_data={“task”: “analyze market trends”})

Memory Module

MemoryManager - Unified interface for all Praval agent memory systems

This coordinates: - Short-term working memory - Long-term vector memory - Episodic conversation memory - Semantic knowledge memory

class praval.memory.memory_manager.MemoryManager(agent_id, backend='auto', qdrant_url='http://localhost:6333', storage_path=None, collection_name='praval_memories', short_term_max_entries=1000, short_term_retention_hours=24, knowledge_base_path=None)[source]

Bases: object

Unified memory management system for Praval agents

Provides a single interface to: - Store and retrieve memories across all systems - Coordinate between short-term and long-term storage - Manage different types of memory (episodic, semantic, etc.) - Optimize memory access patterns

Parameters:

agent_id (str)
backend (str)
qdrant_url (str)
storage_path (str | None)
collection_name (str)
short_term_max_entries (int)
short_term_retention_hours (int)
knowledge_base_path (str | None)

__init__(agent_id, backend='auto', qdrant_url='http://localhost:6333', storage_path=None, collection_name='praval_memories', short_term_max_entries=1000, short_term_retention_hours=24, knowledge_base_path=None)[source]

Initialize the unified memory manager

Parameters:

agent_id (str) – ID of the agent using this memory
backend (str) – Memory backend (“auto”, “chromadb”, “qdrant”, “memory”)
qdrant_url (str) – URL for Qdrant vector database
storage_path (Optional[str]) – Path for persistent storage
collection_name (str) – Collection name for vector storage
short_term_max_entries (int) – Max entries in short-term memory
short_term_retention_hours (int) – Short-term memory retention time
knowledge_base_path (Optional[str]) – Path to knowledge base files to auto-index

store_memory(agent_id, content, memory_type=MemoryType.SHORT_TERM, metadata=None, importance=0.5, store_long_term=None)[source]

Store a memory entry

Parameters:

agent_id (str) – The agent storing the memory
content (str) – The memory content
memory_type (MemoryType) – Type of memory
metadata (Optional[Dict[str, Any]]) – Additional metadata
importance (float) – Importance score (0.0 to 1.0)
store_long_term (bool) – Whether to store in long-term memory (auto-decided if None)

Return type:

Returns:

Memory ID

retrieve_memory(memory_id)[source]

Retrieve a specific memory by ID

Parameters:: memory_id (str) – The memory ID
Return type:: Optional[MemoryEntry]
Returns:: The memory entry if found

search_memories(query)[source]

Search memories across all systems

Parameters:: query (MemoryQuery) – The search query
Return type:: MemorySearchResult
Returns:: Combined search results

get_conversation_context(agent_id, turns=10)[source]

Get recent conversation context for an agent

Parameters:

agent_id (str) – The agent ID
turns (int) – Number of conversation turns

Return type:

Returns:

List of conversation memories

store_conversation_turn(agent_id, user_message, agent_response, context=None)[source]

Store a conversation turn

Parameters:

agent_id (str) – The agent ID
user_message (str) – User’s message
agent_response (str) – Agent’s response
context (Optional[Dict[str, Any]]) – Additional context

Return type:

Returns:

Memory ID

store_knowledge(agent_id, knowledge, domain='general', confidence=1.0, knowledge_type='fact')[source]

Store knowledge or facts

Parameters:

agent_id (str) – The agent ID
knowledge (str) – The knowledge content
domain (str) – Domain of knowledge
confidence (float) – Confidence in the knowledge
knowledge_type (str) – Type of knowledge (fact, concept, rule)

Return type:

Returns:

Memory ID

get_domain_knowledge(agent_id, domain, limit=20)[source]

Get knowledge in a specific domain

Parameters:

agent_id (str) – The agent ID
domain (str) – The domain
limit (int) – Maximum results

Return type:

Returns:

List of knowledge entries

clear_agent_memories(agent_id, memory_types=None)[source]

Clear memories for a specific agent

Parameters:

agent_id (str) – The agent ID
memory_types (Optional[List[MemoryType]]) – Types of memory to clear (all if None)

get_memory_stats()[source]

Get comprehensive memory statistics

Return type:: Dict[str, Any]

health_check()[source]

Check health of all memory systems

Return type:: Dict[str, bool]

recall_by_id(memory_id)[source]

Recall a specific memory by ID (for spore references)

Return type:: List[MemoryEntry]
Parameters:: memory_id (str)

get_knowledge_references(content, importance_threshold=0.7)[source]

Get knowledge references for lightweight spores

Return type:

List[str]

Parameters:

content (str)
importance_threshold (float)

shutdown()[source]: Shutdown all memory systems

Short-term working memory for Praval agents

This provides fast, in-memory storage for: - Current conversation context - Recent agent interactions - Active tasks and goals - Temporary state information

class praval.memory.short_term_memory.ShortTermMemory(max_entries=1000, retention_hours=24, cleanup_interval=3600)[source]

Bases: object

Fast, in-memory storage for short-term agent memory

Features: - Thread-safe operations - Automatic cleanup of old memories - Context-aware retrieval - Working memory capacity limits

Parameters:

max_entries (int)
retention_hours (int)
cleanup_interval (int)

__init__(max_entries=1000, retention_hours=24, cleanup_interval=3600)[source]

Initialize short-term memory

Parameters:

max_entries (int) – Maximum number of entries to keep
retention_hours (int) – How long to keep memories (hours)
cleanup_interval (int) – How often to cleanup old memories (seconds)

store(memory)[source]

Store a memory entry

Parameters:: memory (MemoryEntry) – The memory entry to store
Return type:: str
Returns:: The ID of the stored memory

retrieve(memory_id)[source]

Retrieve a specific memory by ID

Parameters:: memory_id (str) – The ID of the memory to retrieve
Return type:: Optional[MemoryEntry]
Returns:: The memory entry if found, None otherwise

search(query)[source]

Search memories using text similarity

Parameters:: query (MemoryQuery) – The search query
Return type:: MemorySearchResult
Returns:: Search results with matching memories

get_recent(agent_id=None, limit=10)[source]

Get recent memories

Parameters:

agent_id (Optional[str]) – Filter by specific agent
limit (int) – Maximum number of memories to return

Return type:

Returns:

List of recent memory entries

get_context(agent_id, context_size=5)[source]

Get contextual memories for an agent

Parameters:

agent_id (str) – The agent to get context for
context_size (int) – Number of contextual memories

Return type:

Returns:

List of contextual memory entries

clear_agent_memories(agent_id)[source]

Clear all memories for a specific agent

Parameters:: agent_id (str)

get_stats()[source]

Get memory statistics

Return type:: Dict[str, Any]

shutdown()[source]: Shutdown the memory system

Long-term vector memory using Qdrant for Praval agents

This provides persistent, vector-based storage for: - Semantic knowledge and concepts - Long-term conversation history - Learned patterns and insights - Cross-session memory persistence

class praval.memory.long_term_memory.LongTermMemory(qdrant_url='http://localhost:6333', collection_name='praval_memories', vector_size=1536, distance_metric='cosine')[source]

Bases: object

Qdrant-based long-term memory for persistent vector storage

Features: - Vector similarity search - Persistent storage across sessions - Scalable to millions of memories - Semantic search capabilities - Memory importance scoring

Parameters:

qdrant_url (str)
collection_name (str)
vector_size (int)
distance_metric (str)

__init__(qdrant_url='http://localhost:6333', collection_name='praval_memories', vector_size=1536, distance_metric='cosine')[source]

Initialize long-term memory

Parameters:

qdrant_url (str) – URL to Qdrant instance
collection_name (str) – Name of the collection to use
vector_size (int) – Size of embedding vectors
distance_metric (str) – Distance metric for similarity search

store(memory)[source]

Store a memory entry with vector embedding

Parameters:: memory (MemoryEntry) – The memory entry to store
Return type:: str
Returns:: The ID of the stored memory

retrieve(memory_id)[source]

Retrieve a specific memory by ID

Parameters:: memory_id (str) – The ID of the memory to retrieve
Return type:: Optional[MemoryEntry]
Returns:: The memory entry if found, None otherwise

search(query)[source]

Search memories using vector similarity

Parameters:: query (MemoryQuery) – The search query
Return type:: MemorySearchResult
Returns:: Search results with matching memories

delete(memory_id)[source]

Delete a memory entry

Parameters:: memory_id (str) – The ID of the memory to delete
Return type:: bool
Returns:: True if deleted successfully, False otherwise

clear_agent_memories(agent_id)[source]

Clear all memories for a specific agent

Parameters:: agent_id (str)

get_stats()[source]

Get memory statistics

Return type:: Dict[str, Any]

health_check()[source]

Check if Qdrant connection is healthy

Return type:: bool

Episodic memory for Praval agents - conversation history and experiences

This manages: - Conversation turns and dialogue history - Agent interaction sequences - Temporal event chains - Experience-based learning patterns

class praval.memory.episodic_memory.EpisodicMemory(long_term_memory, short_term_memory, conversation_window=50, episode_lifetime_days=30)[source]

Bases: object

Manages episodic memories - experiences and conversations over time

Features: - Conversation turn tracking - Experience sequencing - Temporal relationship modeling - Context window management

Parameters:

long_term_memory (LongTermMemory)
short_term_memory (ShortTermMemory)
conversation_window (int)
episode_lifetime_days (int)

__init__(long_term_memory, short_term_memory, conversation_window=50, episode_lifetime_days=30)[source]

Initialize episodic memory

Parameters:

long_term_memory (LongTermMemory) – Long-term memory backend
short_term_memory (ShortTermMemory) – Short-term memory backend
conversation_window (int) – Number of conversation turns to keep in context
episode_lifetime_days (int) – How long to keep episodes before archiving

store_conversation_turn(agent_id, user_message, agent_response, context=None)[source]

Store a conversation turn as an episodic memory

Parameters:

agent_id (str) – The agent involved in the conversation
user_message (str) – The user’s message
agent_response (str) – The agent’s response
context (Optional[Dict[str, Any]]) – Additional context information

Return type:

Returns:

The memory ID

store_experience(agent_id, experience_type, experience_data, outcome, success=True)[source]

Store an experience or learning episode

Parameters:

agent_id (str) – The agent that had the experience
experience_type (str) – Type of experience (e.g., “task_completion”, “problem_solving”)
experience_data (Dict[str, Any]) – Data about the experience
outcome (str) – The result or outcome
success (bool) – Whether the experience was successful

Return type:

Returns:

The memory ID

get_conversation_context(agent_id, turns=None)[source]

Get recent conversation context for an agent

Parameters:

agent_id (str) – The agent to get context for
turns (Optional[int]) – Number of conversation turns (default: conversation_window)

Return type:

Returns:

List of recent conversation memories

get_similar_experiences(agent_id, experience_description, limit=5)[source]

Find similar past experiences for an agent

Parameters:

agent_id (str) – The agent to search experiences for
experience_description (str) – Description of the current experience
limit (int) – Maximum number of similar experiences to return

Return type:

MemorySearchResult

Returns:

Search results with similar experiences

get_episode_timeline(agent_id, start_time, end_time)[source]

Get episodic memories within a time range

Parameters:

agent_id (str) – The agent to get timeline for
start_time (datetime) – Start of the time range
end_time (datetime) – End of the time range

Return type:

Returns:

List of episodic memories in chronological order

archive_old_episodes(cutoff_days=None)[source]

Archive old episodic memories (implementation depends on requirements)

Parameters:: cutoff_days (Optional[int]) – Days after which to archive (default: episode_lifetime_days)

get_stats()[source]

Get episodic memory statistics

Return type:: Dict[str, Any]

Semantic memory for Praval agents - knowledge, facts, and concepts

This manages: - Factual knowledge storage - Concept relationships - Domain expertise - Learned information persistence

class praval.memory.semantic_memory.SemanticMemory(long_term_memory)[source]

Bases: object

Manages semantic memories - facts, concepts, and knowledge

Features: - Factual knowledge storage - Concept relationship tracking - Knowledge validation and updating - Domain expertise building

Parameters:: long_term_memory (LongTermMemory)

__init__(long_term_memory)[source]

Initialize semantic memory

Parameters:: long_term_memory (LongTermMemory) – Long-term memory backend for persistence

store_fact(agent_id, fact, domain, confidence=1.0, source=None, related_concepts=None)[source]

Store a factual piece of knowledge

Parameters:

agent_id (str) – The agent learning this fact
fact (str) – The factual statement
domain (str) – Domain or category of knowledge
confidence (float) – Confidence in this fact (0.0 to 1.0)
source (Optional[str]) – Source of this information
related_concepts (Optional[List[str]]) – Related concepts or topics

Return type:

Returns:

The memory ID

store_concept(agent_id, concept, definition, domain, properties=None, relationships=None)[source]

Store a concept with its definition and relationships

Parameters:

agent_id (str) – The agent learning this concept
concept (str) – The concept name
definition (str) – Definition of the concept
domain (str) – Domain or field of the concept
properties (Optional[Dict[str, Any]]) – Properties or attributes of the concept
relationships (Optional[Dict[str, List[str]]]) – Relationships to other concepts (e.g., {“is_a”: [“category”], “relates_to”: [“other_concepts”]})

Return type:

Returns:

The memory ID

store_rule(agent_id, rule_name, rule_description, conditions, actions, domain, confidence=1.0)[source]

Store a procedural rule or pattern

Parameters:

agent_id (str) – The agent learning this rule
rule_name (str) – Name of the rule
rule_description (str) – Description of what the rule does
conditions (List[str]) – Conditions when the rule applies
actions (List[str]) – Actions to take when conditions are met
domain (str) – Domain of application
confidence (float) – Confidence in this rule

Return type:

Returns:

The memory ID

get_knowledge_in_domain(agent_id, domain, limit=50)[source]

Get all knowledge in a specific domain

Parameters:

agent_id (str) – The agent to get knowledge for
domain (str) – The domain to search in
limit (int) – Maximum number of entries to return

Return type:

Returns:

List of semantic memories in the domain

find_related_concepts(agent_id, concept, limit=10)[source]

Find concepts related to a given concept

Parameters:

agent_id (str) – The agent to search for
concept (str) – The concept to find relations for
limit (int) – Maximum number of related concepts

Return type:

MemorySearchResult

Returns:

Search results with related concepts

validate_knowledge(agent_id, statement, threshold=0.8)[source]

Check if a statement is consistent with stored knowledge

Parameters:

agent_id (str) – The agent to check knowledge for
statement (str) – The statement to validate
threshold (float) – Similarity threshold for matching

Return type:

Dict[str, Any]

Returns:

Validation result with confidence and supporting evidence

update_knowledge(agent_id, old_knowledge, new_knowledge, reason='Updated information')[source]

Update existing knowledge with new information

Parameters:

agent_id (str) – The agent updating knowledge
old_knowledge (str) – The knowledge to update
new_knowledge (str) – The new knowledge
reason (str) – Reason for the update

Return type:

Returns:

True if update was successful

get_domain_expertise_level(agent_id, domain)[source]

Assess the agent’s expertise level in a domain

Parameters:

agent_id (str) – The agent to assess
domain (str) – The domain to assess expertise in

Return type:

Dict[str, Any]

Returns:

Expertise assessment with metrics

get_stats()[source]

Get semantic memory statistics

Return type:: Dict[str, Any]

Memory types and data structures for Praval agents

class praval.memory.memory_types.MemoryType(*values)[source]

Bases: Enum

Types of memory entries

SHORT_TERM = 'short_term'

EPISODIC = 'episodic'

SEMANTIC = 'semantic'

PROCEDURAL = 'procedural'

EMOTIONAL = 'emotional'

class praval.memory.memory_types.MemoryEntry(id, agent_id, memory_type, content, metadata, embedding=None, created_at=None, accessed_at=None, access_count=0, importance=0.5)[source]

Bases: object

A single memory entry

Parameters:

id (str)
agent_id (str)
memory_type (MemoryType)
content (str)
metadata (Dict[str, Any])
embedding (List[float] | None)
created_at (datetime)
accessed_at (datetime)
access_count (int)
importance (float)

id: str

agent_id: str

memory_type: MemoryType

content: str

metadata: Dict[str, Any]

embedding: Optional[List[float]] = None

created_at: datetime = None

accessed_at: datetime = None

access_count: int = 0

importance: float = 0.5

mark_accessed()[source]: Mark this memory as accessed

to_dict()[source]

Convert to dictionary for storage

Return type:: Dict[str, Any]

classmethod from_dict(data)[source]

Create from dictionary

Return type:: MemoryEntry
Parameters:: data (Dict[str, Any])

__init__(id, agent_id, memory_type, content, metadata, embedding=None, created_at=None, accessed_at=None, access_count=0, importance=0.5)

Parameters:

id (str)
agent_id (str)
memory_type (MemoryType)
content (str)
metadata (Dict[str, Any])
embedding (List[float] | None)
created_at (datetime)
accessed_at (datetime)
access_count (int)
importance (float)

Return type:

None

class praval.memory.memory_types.MemoryQuery(query_text, memory_types=None, agent_id=None, limit=10, similarity_threshold=0.7, include_metadata=True, temporal_filter=None)[source]

Bases: object

A memory search query

Parameters:

query_text (str)
memory_types (List[MemoryType])
agent_id (str | None)
limit (int)
similarity_threshold (float)
include_metadata (bool)
temporal_filter (Dict[str, datetime] | None)

query_text: str

memory_types: List[MemoryType] = None

agent_id: Optional[str] = None

limit: int = 10

similarity_threshold: float = 0.7

include_metadata: bool = True

temporal_filter: Optional[Dict[str, datetime]] = None

__init__(query_text, memory_types=None, agent_id=None, limit=10, similarity_threshold=0.7, include_metadata=True, temporal_filter=None)

Parameters:

query_text (str)
memory_types (List[MemoryType])
agent_id (str | None)
limit (int)
similarity_threshold (float)
include_metadata (bool)
temporal_filter (Dict[str, datetime] | None)

Return type:

None

class praval.memory.memory_types.MemorySearchResult(entries, scores, query, total_found)[source]

Bases: object

Result of a memory search

Parameters:

entries (List[MemoryEntry])
scores (List[float])
query (MemoryQuery)
total_found (int)

entries: List[MemoryEntry]

scores: List[float]

query: MemoryQuery

total_found: int

get_best_match()[source]

Get the best matching memory entry

Return type:: Optional[MemoryEntry]

get_above_threshold(threshold=0.8)[source]

Get entries above similarity threshold

Return type:: List[MemoryEntry]
Parameters:: threshold (float)

__init__(entries, scores, query, total_found)

Parameters:

entries (List[MemoryEntry])
scores (List[float])
query (MemoryQuery)
total_found (int)

Return type:

None

Storage Module

Data Manager - Unified interface for storage operations

Provides a high-level interface for agents to interact with multiple storage providers through a single, consistent API.

class praval.storage.data_manager.DataManager(registry=None)[source]

Bases: object

Unified data management interface for agents.

Provides high-level methods for storing, retrieving, and querying data across multiple storage backends through a single interface.

Features: - Automatic provider selection based on data type - Cross-provider queries and operations - Data reference management for spore communication - Storage optimization and caching - Transaction-like operations across providers

Parameters:: registry (StorageRegistry | None)

__init__(registry=None)[source]

Initialize data manager.

Parameters:: registry (Optional[StorageRegistry]) – Storage registry to use (defaults to global registry)

set_agent_context(agent_name)[source]

Set the current agent context for permission checking.

Parameters:: agent_name (str)

get_agent_context()[source]

Get the current agent context.

Return type:: Optional[str]

async store(provider, resource, data, **kwargs)[source]

Store data in a specific provider.

Parameters:

provider (str) – Storage provider name
resource (str) – Resource identifier (table, key, path, etc.)
data (Any) – Data to store
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with operation outcome

async get(provider, resource, **kwargs)[source]

Retrieve data from a specific provider.

Parameters:

provider (str) – Storage provider name
resource (str) – Resource identifier
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with retrieved data

async query(provider, resource, query, **kwargs)[source]

Execute a query on a specific provider.

Parameters:

provider (str) – Storage provider name
resource (str) – Resource to query
query (Union[str, Dict]) – Query string or structured query
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(provider, resource, **kwargs)[source]

Delete data from a specific provider.

Parameters:

provider (str) – Storage provider name
resource (str) – Resource to delete
**kwargs – Delete parameters

Return type:

Returns:

StorageResult with operation outcome

async smart_store(data, resource=None, preferred_provider=None, **kwargs)[source]

Intelligently store data by selecting the best provider.

Parameters:

data (Any) – Data to store
resource (Optional[str]) – Optional resource identifier
preferred_provider (Optional[str]) – Preferred storage provider
**kwargs – Storage parameters

Return type:

Returns:

StorageResult with operation outcome

async smart_search(query, providers=None, **kwargs)[source]

Search across multiple providers intelligently.

Parameters:

query (Union[str, List[float], Dict]) – Search query (text, vector, or structured)
providers (Optional[List[str]]) – Providers to search (defaults to all suitable)
**kwargs – Search parameters

Return type:

List[StorageResult]

Returns:

List of StorageResult from different providers

create_data_reference(provider, resource, **metadata)[source]

Create a data reference for spore communication.

Parameters:

provider (str) – Storage provider name
resource (str) – Resource identifier
**metadata – Additional metadata

Return type:

DataReference

Returns:

DataReference object

async resolve_data_reference(data_ref, **kwargs)[source]

Resolve a data reference to actual data.

Parameters:

data_ref (Union[DataReference, str]) – DataReference object or URI string
**kwargs – Retrieval parameters

Return type:

Returns:

StorageResult with resolved data

async batch_store(operations)[source]

Execute multiple store operations in batch.

Parameters:: operations (List[Dict[str, Any]]) – List of operation dictionaries with keys: - provider: Storage provider name - resource: Resource identifier - data: Data to store - kwargs: Additional parameters
Return type:: List[StorageResult]
Returns:: List of StorageResult objects

async batch_get(operations)[source]

Execute multiple get operations in batch.

Parameters:: operations (List[Dict[str, Any]]) – List of operation dictionaries with keys: - provider: Storage provider name - resource: Resource identifier - kwargs: Additional parameters
Return type:: List[StorageResult]
Returns:: List of StorageResult objects

list_providers(storage_type=None)[source]

List available storage providers.

Parameters:: storage_type (Optional[str]) – Filter by storage type
Return type:: List[str]
Returns:: List of provider names

get_provider_info(provider)[source]

Get information about a storage provider.

Parameters:: provider (str) – Storage provider name
Return type:: Dict[str, Any]
Returns:: Provider information dictionary

async health_check(provider=None)[source]

Perform health check on providers.

Parameters:: provider (Optional[str]) – Specific provider to check (None for all)
Return type:: Dict[str, Any]
Returns:: Health check results

praval.storage.data_manager.get_data_manager()[source]

Get the global data manager instance.

Return type:: DataManager

async praval.storage.data_manager.store_data(provider, resource, data, **kwargs)[source]

Store data using the global data manager.

Return type:

Parameters:

provider (str)
resource (str)
data (Any)

async praval.storage.data_manager.get_data(provider, resource, **kwargs)[source]

Retrieve data using the global data manager.

Return type:

Parameters:

provider (str)
resource (str)

async praval.storage.data_manager.query_data(provider, resource, query, **kwargs)[source]

Query data using the global data manager.

Return type:

Parameters:

provider (str)
resource (str)
query (str | Dict)

async praval.storage.data_manager.delete_data(provider, resource, **kwargs)[source]

Delete data using the global data manager.

Return type:

Parameters:

provider (str)
resource (str)

Base Storage Provider Framework

Defines the core interfaces and base classes for all Praval storage providers. This provides a standardized way to create, register, and use storage backends that agents can access uniformly.

class praval.storage.base_provider.StorageType(*values)[source]

Bases: Enum

Types of storage backends

RELATIONAL = 'relational'

DOCUMENT = 'document'

KEY_VALUE = 'key_value'

OBJECT = 'object'

VECTOR = 'vector'

SEARCH = 'search'

GRAPH = 'graph'

FILE_SYSTEM = 'file_system'

CACHE = 'cache'

QUEUE = 'queue'

class praval.storage.base_provider.DataReference(provider, storage_type, resource_id, metadata=<factory>, created_at=<factory>, expires_at=None)[source]

Bases: object

Reference to data stored in a backend

Parameters:

provider (str)
storage_type (StorageType)
resource_id (str)
metadata (Dict[str, Any])
created_at (datetime)
expires_at (datetime | None)

provider: str

storage_type: StorageType

resource_id: str

metadata: Dict[str, Any]

created_at: datetime

expires_at: Optional[datetime] = None

to_uri()[source]

Convert to URI format for spore communication

Return type:: str

classmethod from_uri(uri)[source]

Create DataReference from URI

Return type:: DataReference
Parameters:: uri (str)

is_expired()[source]

Check if reference has expired

Return type:: bool

__init__(provider, storage_type, resource_id, metadata=<factory>, created_at=<factory>, expires_at=None)

Parameters:

provider (str)
storage_type (StorageType)
resource_id (str)
metadata (Dict[str, Any])
created_at (datetime)
expires_at (datetime | None)

Return type:

None

class praval.storage.base_provider.StorageQuery(operation, resource, parameters=<factory>, filters=<factory>, limit=None, offset=None, timeout=None)[source]

Bases: object

Query parameters for storage operations

Parameters:

operation (str)
resource (str)
parameters (Dict[str, Any])
filters (Dict[str, Any])
limit (int | None)
offset (int | None)
timeout (float | None)

operation: str

resource: str

parameters: Dict[str, Any]

filters: Dict[str, Any]

limit: Optional[int] = None

offset: Optional[int] = None

timeout: Optional[float] = None

__init__(operation, resource, parameters=<factory>, filters=<factory>, limit=None, offset=None, timeout=None)

Parameters:

operation (str)
resource (str)
parameters (Dict[str, Any])
filters (Dict[str, Any])
limit (int | None)
offset (int | None)
timeout (float | None)

Return type:

None

class praval.storage.base_provider.StorageResult(success, data=None, error=None, execution_time=0.0, metadata=<factory>, data_reference=None, timestamp=<factory>)[source]

Bases: object

Result from storage operation

Parameters:

success (bool)
data (Any)
error (str | None)
execution_time (float)
metadata (Dict[str, Any])
data_reference (DataReference | None)
timestamp (datetime)

success: bool

data: Any = None

error: Optional[str] = None

execution_time: float = 0.0

metadata: Dict[str, Any]

data_reference: Optional[DataReference] = None

timestamp: datetime

__init__(success, data=None, error=None, execution_time=0.0, metadata=<factory>, data_reference=None, timestamp=<factory>)

Parameters:

success (bool)
data (Any)
error (str | None)
execution_time (float)
metadata (Dict[str, Any])
data_reference (DataReference | None)
timestamp (datetime)

Return type:

None

class praval.storage.base_provider.StorageMetadata(name, description, storage_type, version='1.0.0', supports_async=True, supports_transactions=False, supports_schemas=False, supports_indexing=False, supports_search=False, supports_streaming=False, max_connection_pool=10, default_timeout=30.0, required_config=<factory>, optional_config=<factory>, connection_string_template=None)[source]

Bases: object

Metadata describing a storage provider’s capabilities

Parameters:

name (str)
description (str)
storage_type (StorageType)
version (str)
supports_async (bool)
supports_transactions (bool)
supports_schemas (bool)
supports_indexing (bool)
supports_search (bool)
supports_streaming (bool)
max_connection_pool (int)
default_timeout (float)
required_config (List[str])
optional_config (List[str])
connection_string_template (str | None)

name: str

description: str

storage_type: StorageType

version: str = '1.0.0'

supports_async: bool = True

supports_transactions: bool = False

supports_schemas: bool = False

supports_indexing: bool = False

supports_search: bool = False

supports_streaming: bool = False

max_connection_pool: int = 10

default_timeout: float = 30.0

required_config: List[str]

optional_config: List[str]

connection_string_template: Optional[str] = None

__init__(name, description, storage_type, version='1.0.0', supports_async=True, supports_transactions=False, supports_schemas=False, supports_indexing=False, supports_search=False, supports_streaming=False, max_connection_pool=10, default_timeout=30.0, required_config=<factory>, optional_config=<factory>, connection_string_template=None)

Parameters:

name (str)
description (str)
storage_type (StorageType)
version (str)
supports_async (bool)
supports_transactions (bool)
supports_schemas (bool)
supports_indexing (bool)
supports_search (bool)
supports_streaming (bool)
max_connection_pool (int)
default_timeout (float)
required_config (List[str])
optional_config (List[str])
connection_string_template (str | None)

Return type:

None

class praval.storage.base_provider.BaseStorageProvider(name, config)[source]

Bases: ABC

Abstract base class for all storage providers.

All storage backends must inherit from this class and implement the required methods. This ensures a consistent interface across all storage types while allowing for provider-specific optimizations.

Parameters:

name (str)
config (Dict[str, Any])

__init__(name, config)[source]

Initialize the storage provider.

Parameters:

name (str) – Unique name for this provider instance
config (Dict[str, Any]) – Provider-specific configuration

abstractmethod async connect()[source]

Establish connection to the storage backend.

Return type:: bool
Returns:: True if connection successful, False otherwise

abstractmethod async disconnect()[source]: Close connection to the storage backend.

abstractmethod async store(resource, data, **kwargs)[source]

Store data in the backend.

Parameters:

resource (str) – Resource identifier (table, bucket, key, etc.)
data (Any) – Data to store
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with operation outcome

abstractmethod async retrieve(resource, **kwargs)[source]

Retrieve data from the backend.

Parameters:

resource (str) – Resource identifier
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with retrieved data

abstractmethod async query(resource, query, **kwargs)[source]

Execute a query against the backend.

Parameters:

resource (str) – Resource to query
query (Union[str, Dict]) – Query string or structured query
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with query results

abstractmethod async delete(resource, **kwargs)[source]

Delete data from the backend.

Parameters:

resource (str) – Resource to delete
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with operation outcome

async exists(resource, **kwargs)[source]

Check if a resource exists.

Parameters:

resource (str) – Resource to check
**kwargs – Provider-specific parameters

Return type:

Returns:

True if resource exists, False otherwise

async list_resources(prefix='', **kwargs)[source]

List available resources.

Parameters:

prefix (str) – Resource prefix to filter by
**kwargs – Provider-specific parameters

Return type:

Returns:

StorageResult with list of resources

async safe_execute(operation, *args, **kwargs)[source]

Execute operation with error handling and timing.

Parameters:

operation (str) – Operation name
*args – Operation parameters
**kwargs –
Operation parameters

Return type:

Returns:

StorageResult with operation outcome

get_schema()[source]

Get provider schema/capabilities.

Return type:: Dict[str, Any]

async health_check()[source]

Perform health check on the storage backend.

Return type:: Dict[str, Any]

praval.storage.base_provider.create_storage_provider(provider_class, name, config)[source]

Create a storage provider instance.

Parameters:

provider_class (Type[BaseStorageProvider]) – Provider class to instantiate
name (str) – Provider instance name
config (Dict[str, Any]) – Provider configuration

Return type:

BaseStorageProvider

Returns:

Configured provider instance

Storage Registry System

Provides centralized registration, discovery, and management of storage providers available to Praval agents. This follows the same pattern as the tool registry for consistent framework design.

class praval.storage.storage_registry.StorageRegistry[source]

Bases: object

Central registry for managing storage providers available to agents.

Features: - Provider registration and discovery - Type-based organization (relational, object, vector, etc.) - Access control and permissions - Connection pooling and lifecycle management - Usage statistics and health monitoring - Multi-provider query routing

__init__()[source]

async register_provider(provider, replace_existing=False, permissions=None, auto_connect=None)[source]

Register a storage provider in the registry.

Parameters:

provider (BaseStorageProvider) – Provider instance to register
replace_existing (bool) – Whether to replace existing provider with same name
permissions (Optional[List[str]]) – List of agent names allowed to use this provider
auto_connect (bool) – Whether to auto-connect the provider

Return type:

Returns:

True if registration successful, False otherwise

async unregister_provider(provider_name)[source]

Unregister a provider from the registry.

Parameters:: provider_name (str) – Name of provider to unregister
Return type:: bool
Returns:: True if unregistration successful, False if provider not found

get_provider(provider_name, agent_name=None)[source]

Get a provider by name with permission checking.

Parameters:

provider_name (str) – Name of the provider to retrieve
agent_name (str) – Name of the agent requesting the provider

Return type:

BaseStorageProvider

Returns:

BaseStorageProvider instance

Raises:

StorageNotFoundError – If provider not found
StoragePermissionError – If agent lacks permission

list_providers(storage_type=None, agent_name=None, connected_only=False)[source]

List available providers with optional filtering.

Parameters:

storage_type (Optional[StorageType]) – Filter by storage type
agent_name (Optional[str]) – Filter by agent permissions
connected_only (bool) – Only return connected providers

Return type:

List[str]

Returns:

List of provider names

get_providers_by_type(storage_type)[source]

Get all providers of a specific storage type.

Return type:: List[str]
Parameters:: storage_type (StorageType)

get_storage_types()[source]

Get list of all available storage types.

Return type:: List[StorageType]

async execute_query(provider_name, query, agent_name=None)[source]

Execute a query on a specific provider.

Parameters:

provider_name (str) – Name of provider to query
query (StorageQuery) – Query to execute
agent_name (Optional[str]) – Name of requesting agent

Return type:

Returns:

StorageResult with query outcome

async health_check_all()[source]

Perform health checks on all registered providers.

Return type:: Dict[str, Dict[str, Any]]

get_usage_stats(provider_name=None)[source]

Get usage statistics for providers.

Return type:: Union[Dict[str, Any], Dict[str, Dict[str, Any]]]
Parameters:: provider_name (str | None)

get_registry_info()[source]

Get comprehensive information about the registry.

Return type:: Dict[str, Any]

set_permissions(provider_name, agent_names)[source]

Set permissions for a provider.

Parameters:

provider_name (str)
agent_names (List[str])

add_permission(provider_name, agent_name)[source]

Add permission for an agent to use a provider.

Parameters:

provider_name (str)
agent_name (str)

remove_permission(provider_name, agent_name)[source]

Remove permission for an agent to use a provider.

Parameters:

provider_name (str)
agent_name (str)

block_provider(provider_name)[source]

Block a provider from being used.

Parameters:: provider_name (str)

unblock_provider(provider_name)[source]

Unblock a previously blocked provider.

Parameters:: provider_name (str)

async cleanup_registry()[source]: Clean up disconnected providers and expired data.

praval.storage.storage_registry.get_storage_registry()[source]

Get the global storage registry instance.

Return type:: StorageRegistry

async praval.storage.storage_registry.register_storage_provider(provider, **kwargs)[source]

Register a storage provider in the global registry.

Return type:: bool
Parameters:: provider (BaseStorageProvider)

praval.storage.storage_registry.get_storage_provider(provider_name, agent_name=None)[source]

Get a storage provider from the global registry.

Return type:

BaseStorageProvider

Parameters:

provider_name (str)
agent_name (str)

praval.storage.storage_registry.list_storage_providers(**kwargs)[source]

List available storage providers from the global registry.

Return type:: List[str]

Storage decorators for Praval agents

Provides decorator-based integration between agents and storage providers, following the same patterns as the agent and tool decorators.

praval.storage.decorators.storage_enabled(providers=None, auto_register=True, permissions=None, **default_configs)[source]

Decorator to enable storage access for agent functions.

Parameters:

providers (Union[str, List[str], Dict[str, Dict[str, Any]], None]) – Storage providers to enable. Can be: - String: Single provider name - List: Multiple provider names - Dict: Provider name -> configuration mapping
auto_register (bool) – Whether to auto-register providers from environment
permissions (Optional[List[str]]) – Default permissions for storage access
**default_configs – Default configurations for providers

Examples

@storage_enabled(“postgres”) @agent(“data_analyst”) def analyze_data(spore):

data = storage.query(“postgres”, “SELECT * FROM customers”)

@storage_enabled([“postgres”, “s3”, “redis”]) @agent(“business_intelligence”) def generate_report(spore):

# Access multiple storage backends pass

@storage_enabled({: “postgres”: {“host”: “localhost”, “database”: “business”}, “s3”: {“bucket_name”: “reports”}

}) @agent(“report_generator”) def create_analysis(spore):

pass

praval.storage.decorators.requires_storage(*provider_names, permissions=None)[source]

Decorator to require specific storage providers for a function.

Parameters:

*provider_names (str) – Names of required storage providers
permissions (Optional[List[str]]) – Required permissions for storage access

Example

@requires_storage(“postgres”, “s3”) @agent(“data_processor”) def process_customer_data(spore):

# Function requires both postgres and s3 to be available pass

Storage Providers

File system storage provider for Praval framework

Provides local file system storage capabilities with path management, directory operations, and file metadata support.

class praval.storage.providers.filesystem.FileSystemProvider(name, config)[source]

Local file system storage provider.

Features: - File and directory operations - Path management and validation - File metadata and permissions - Recursive operations - Pattern-based file listing - Atomic file operations

Parameters:

name (str)
config (Dict[str, Any])

async connect()[source]

Verify file system access.

Return type:: bool

async disconnect()[source]: No explicit disconnection needed for file system.

async store(resource, data, **kwargs)[source]

Store data to file system.

Parameters:

resource (str) – File path relative to base_path
data (Any) – Data to store (string, bytes, dict/list for JSON, or file-like object)
**kwargs – File parameters (encoding, mode, etc.)

Return type:

Returns:

StorageResult with operation outcome

async retrieve(resource, **kwargs)[source]

Retrieve data from file system.

Parameters:

resource (str) – File path relative to base_path
**kwargs – Read parameters (encoding, decode_json, etc.)

Return type:

Returns:

StorageResult with retrieved data

async query(resource, query, **kwargs)[source]

Execute file system operations.

Parameters:

resource (str) – Path or pattern
query (Union[str, Dict]) – Query type (“list”, “find”, “metadata”, etc.)
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(resource, **kwargs)[source]

Delete file or directory from file system.

Parameters:

resource (str) – Path to delete
**kwargs – Delete parameters (recursive, etc.)

Return type:

Returns:

StorageResult with operation outcome

async list_resources(prefix='', **kwargs)[source]

List files and directories.

Return type:: StorageResult
Parameters:: prefix (str)

PostgreSQL storage provider for Praval framework

Provides relational database capabilities with SQL query support, transactions, and schema management.

class praval.storage.providers.postgresql.PostgreSQLProvider(name, config)[source]

PostgreSQL storage provider with async connection pooling.

Features: - Async connection pooling - SQL query execution - Transaction support - Schema management - JSON column support - Full-text search capabilities

Parameters:

name (str)
config (Dict[str, Any])

async connect()[source]

Establish connection pool to PostgreSQL.

Return type:: bool

async disconnect()[source]: Close connection pool.

async store(resource, data, **kwargs)[source]

Store data in PostgreSQL table.

Parameters:

resource (str) – Table name
data (Any) – Data to store (dict or list of dicts)
**kwargs – Additional parameters (upsert, returning, etc.)

Return type:

Returns:

StorageResult with operation outcome

async retrieve(resource, **kwargs)[source]

Retrieve data from PostgreSQL table.

Parameters:

resource (str) – Table name
**kwargs – Query parameters (where, limit, offset, order_by, etc.)

Return type:

Returns:

StorageResult with retrieved data

async query(resource, query, **kwargs)[source]

Execute SQL query against PostgreSQL.

Parameters:

resource (str) – Table name (ignored for raw SQL)
query (Union[str, Dict]) – SQL query string or structured query dict
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(resource, **kwargs)[source]

Delete data from PostgreSQL table.

Parameters:

resource (str) – Table name
**kwargs – Delete parameters (where clause required)

Return type:

Returns:

StorageResult with operation outcome

async list_resources(prefix='', **kwargs)[source]

List tables in the database.

Return type:: StorageResult
Parameters:: prefix (str)

Redis storage provider for Praval framework

Provides key-value storage, caching, and pub/sub capabilities with Redis backend.

class praval.storage.providers.redis_provider.RedisProvider(name, config)[source]

Redis key-value storage provider with async support.

Features: - Key-value operations (GET, SET, DEL) - Hash operations (HGET, HSET, HGETALL) - List operations (LPUSH, RPUSH, LRANGE) - Set operations (SADD, SMEMBERS) - Expiration and TTL management - Pub/Sub messaging - Lua script execution

Parameters:

name (str)
config (Dict[str, Any])

async connect()[source]

Establish connection to Redis.

Return type:: bool

async disconnect()[source]: Close Redis connection.

async store(resource, data, **kwargs)[source]

Store data in Redis.

Parameters:

resource (str) – Redis key
data (Any) – Data to store (will be JSON serialized if not string)
**kwargs – Redis parameters (ex, px, nx, xx, etc.)

Return type:

Returns:

StorageResult with operation outcome

async retrieve(resource, **kwargs)[source]

Retrieve data from Redis.

Parameters:

resource (str) – Redis key
**kwargs – Additional parameters (decode_json, etc.)

Return type:

Returns:

StorageResult with retrieved data

async query(resource, query, **kwargs)[source]

Execute Redis operations or search queries.

Parameters:

resource (str) – Pattern or specific key
query (Union[str, Dict]) – Query type or Redis command
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(resource, **kwargs)[source]

Delete keys from Redis.

Parameters:

resource (str) – Key or pattern to delete
**kwargs – Delete parameters

Return type:

Returns:

StorageResult with operation outcome

async list_resources(prefix='', **kwargs)[source]

List keys in Redis.

Return type:: StorageResult
Parameters:: prefix (str)

S3 object storage provider for Praval framework

Provides object storage capabilities with S3-compatible backends including AWS S3, MinIO, and other S3-compatible services.

class praval.storage.providers.s3_provider.S3Provider(name, config)[source]

S3-compatible object storage provider.

Features: - Object upload, download, and deletion - Bucket management - Presigned URLs for secure access - Metadata and tagging support - Multipart uploads for large files - Lifecycle management - Cross-region replication support

Parameters:

name (str)
config (Dict[str, Any])

async connect()[source]

Establish connection to S3.

Return type:: bool

async disconnect()[source]: Close S3 connection.

async store(resource, data, **kwargs)[source]

Store object in S3.

Parameters:

resource (str) – S3 object key
data (Any) – Data to store (bytes, string, file-like object, or dict/list for JSON)
**kwargs – S3 parameters (content_type, metadata, acl, etc.)

Return type:

Returns:

StorageResult with operation outcome

async retrieve(resource, **kwargs)[source]

Retrieve object from S3.

Parameters:

resource (str) – S3 object key
**kwargs – Retrieval parameters (range, decode_json, etc.)

Return type:

Returns:

StorageResult with retrieved data

async query(resource, query, **kwargs)[source]

Execute S3 operations or list objects.

Parameters:

resource (str) – Prefix or specific key
query (Union[str, Dict]) – Query type (“list”, “search”, “metadata”, etc.)
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(resource, **kwargs)[source]

Delete object(s) from S3.

Parameters:

resource (str) – Object key or prefix
**kwargs – Delete parameters (recursive, etc.)

Return type:

Returns:

StorageResult with operation outcome

async list_resources(prefix='', **kwargs)[source]

List objects in S3 bucket.

Return type:: StorageResult
Parameters:: prefix (str)

Qdrant vector storage provider for Praval framework

Integrates with existing Praval memory system to provide vector storage capabilities through the storage framework.

class praval.storage.providers.qdrant_provider.QdrantProvider(name, config)[source]

Qdrant vector database storage provider.

Features: - Vector similarity search - Collection management - Point insertion and retrieval - Filtering and metadata search - Batch operations - Integration with Praval memory system

Parameters:

name (str)
config (Dict[str, Any])

async connect()[source]

Establish connection to Qdrant.

Return type:: bool

async disconnect()[source]: Close Qdrant connection.

async store(resource, data, **kwargs)[source]

Store vector data in Qdrant.

Parameters:

resource (str) – Collection name (optional, uses default if not specified)
data (Any) – Data to store - can be: - Single point: {“id”: “…”, “vector”: […], “payload”: {…}} - Multiple points: [{“id”: “…”, “vector”: […], “payload”: {…}}, …] - Just vector: [0.1, 0.2, …] (will generate ID)
**kwargs – Additional parameters

Return type:

Returns:

StorageResult with operation outcome

async retrieve(resource, **kwargs)[source]

Retrieve vectors from Qdrant.

Parameters:

resource (str) – Collection name or “collection:point_id”
**kwargs – Retrieval parameters (point_ids, with_vectors, with_payload)

Return type:

Returns:

StorageResult with retrieved data

async query(resource, query, **kwargs)[source]

Execute vector search or other operations.

Parameters:

resource (str) – Collection name
query (Union[str, Dict]) – Query type or search vector
**kwargs – Query parameters

Return type:

Returns:

StorageResult with query results

async delete(resource, **kwargs)[source]

Delete points from Qdrant.

Parameters:

resource (str) – Collection name or “collection:point_id”
**kwargs – Delete parameters (point_ids, filter)

Return type:

Returns:

StorageResult with operation outcome

async list_resources(prefix='', **kwargs)[source]

List collections in Qdrant.

Return type:: StorageResult
Parameters:: prefix (str)

Tool Module

Tool decorator and utilities for Praval Framework.

This module provides the @tool decorator for creating tools that can be registered and used by agents. Tools are automatically registered in the global tool registry and can be associated with specific agents.

praval.tools.tool(tool_name=None, owned_by=None, description=None, category='general', shared=False, version='1.0.0', author='', tags=None)[source]

Decorator to register a function as a tool in the Praval framework.

The @tool decorator automatically registers functions as tools that can be used by agents. Tools can be owned by specific agents, shared across all agents, or organized by category.

Parameters:

tool_name (Optional[str]) – Name of the tool (defaults to function name)
owned_by (Optional[str]) – Agent that owns this tool
description (Optional[str]) – Description of what the tool does (defaults to docstring)
category (str) – Category for organizing tools
shared (bool) – Whether this tool is available to all agents
version (str) – Version of the tool
author (str) – Author of the tool
tags (Optional[List[str]]) – Tags for tool discovery

Return type:

Callable

Returns:

Decorated function with tool metadata attached

Raises:

ToolError – If tool registration fails or validation errors occur

Examples

Basic tool owned by specific agent: ```python @tool(“add_numbers”, owned_by=”calculator”) def add(x: float, y: float) -> float:

‘’’Add two numbers together.’’’ return x + y

```

Shared tool available to all agents: ```python @tool(“logger”, shared=True, category=”utility”) def log_message(level: str, message: str) -> str:

‘’’Log a message at the specified level.’’’ import logging logger = logging.getLogger(“praval.tools”) getattr(logger, level.lower())(message) return f”Logged: {message}”

```

Tool with metadata: ```python @tool(

“validate_email”, owned_by=”data_processor”, category=”validation”, tags=[“email”, “validation”, “data”], version=”2.0.0”, author=”Praval Team”

) def validate_email(email: str) -> bool:

‘’’Validate email address format.’’’ import re pattern = r’^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+.[a-zA-Z]{2,}$’ return bool(re.match(pattern, email))

```

praval.tools.get_tool_info(tool_func)[source]

Get information about a @tool decorated function.

Parameters:: tool_func (Callable) – Function decorated with @tool
Return type:: dict
Returns:: Dictionary with tool metadata
Raises:: ValueError – If function is not decorated with @tool

praval.tools.is_tool(func)[source]

Check if a function is decorated with @tool.

Parameters:: func (Callable) – Function to check
Return type:: bool
Returns:: True if function is a tool, False otherwise

praval.tools.discover_tools(module=None, pattern=None, category=None)[source]

Discover tools based on various criteria.

Parameters:

module (Optional[str]) – Module name to search for tools
pattern (Optional[str]) – File pattern to search (e.g., “*_tool.py”)
category (Optional[str]) – Category to filter by

Return type:

List[Tool]

Returns:

List of discovered Tool instances

praval.tools.list_tools(agent_name=None, category=None, shared_only=False)[source]

List tools with optional filtering.

Parameters:

agent_name (Optional[str]) – Filter by agent owner
category (Optional[str]) – Filter by category
shared_only (bool) – Only show shared tools

Return type:

List[dict]

Returns:

List of tool information dictionaries

praval.tools.register_tool_with_agent(tool_name, agent_name)[source]

Register an existing tool with an agent at runtime.

Parameters:

tool_name (str) – Name of the tool to register
agent_name (str) – Name of the agent to register with

Return type:

Returns:

True if registration successful, False otherwise

praval.tools.unregister_tool_from_agent(tool_name, agent_name)[source]

Unregister a tool from an agent at runtime.

Parameters:

tool_name (str) – Name of the tool to unregister
agent_name (str) – Name of the agent to unregister from

Return type:

Returns:

True if unregistration successful, False otherwise

class praval.tools.ToolCollection(name, description='')[source]

Bases: object

A collection of related tools that can be managed as a group.

Useful for organizing tools by functionality or creating tool suites that can be easily assigned to agents.

Parameters:

name (str)
description (str)

__init__(name, description='')[source]

Initialize a tool collection.

Parameters:

name (str) – Name of the collection
description (str) – Description of the collection

add_tool(tool_name)[source]

Add a tool to the collection.

Parameters:: tool_name (str) – Name of the tool to add
Raises:: ToolError – If tool doesn’t exist
Return type:: None

remove_tool(tool_name)[source]

Remove a tool from the collection.

Parameters:: tool_name (str) – Name of the tool to remove
Return type:: bool
Returns:: True if removal successful, False if tool wasn’t in collection

assign_to_agent(agent_name)[source]

Assign all tools in the collection to an agent.

Parameters:: agent_name (str) – Name of the agent to assign tools to
Return type:: int
Returns:: Number of tools successfully assigned

get_tools()[source]

Get all tools in the collection.

Return type:: List[Tool]
Returns:: List of Tool instances in the collection

Tool Registry for Praval Framework.

This module provides a centralized registry for managing tools and their relationships to agents. Tools can be registered, discovered, and assigned to agents dynamically.

class praval.core.tool_registry.ToolMetadata(tool_name, owned_by=None, description='', category='general', shared=False, version='1.0.0', author='', tags=<factory>, parameters=<factory>, return_type='Any')[source]

Bases: object

Metadata for a registered tool.

Parameters:

tool_name (str)
owned_by (str | None)
description (str)
category (str)
shared (bool)
version (str)
author (str)
tags (List[str])
parameters (Dict[str, Any])
return_type (str)

tool_name: str

owned_by: Optional[str] = None

description: str = ''

category: str = 'general'

shared: bool = False

version: str = '1.0.0'

author: str = ''

tags: List[str]

parameters: Dict[str, Any]

return_type: str = 'Any'

__init__(tool_name, owned_by=None, description='', category='general', shared=False, version='1.0.0', author='', tags=<factory>, parameters=<factory>, return_type='Any')

Parameters:

tool_name (str)
owned_by (str | None)
description (str)
category (str)
shared (bool)
version (str)
author (str)
tags (List[str])
parameters (Dict[str, Any])
return_type (str)

Return type:

None

class praval.core.tool_registry.Tool(func, metadata)[source]

Bases: object

Wrapper class for a registered tool function.

Provides metadata, validation, and execution capabilities for functions registered as tools in the Praval framework.

Parameters:

func (Callable)
metadata (ToolMetadata)

__init__(func, metadata)[source]

Initialize a Tool instance.

Parameters:

func (Callable) – The function to wrap as a tool
metadata (ToolMetadata) – Metadata describing the tool

Raises:

ToolError – If function validation fails

execute(*args, **kwargs)[source]

Execute the tool function with given arguments.

Parameters:

*args – Positional arguments for the tool function
**kwargs – Keyword arguments for the tool function

Return type:

Any

Returns:

Result of tool function execution

Raises:

ToolError – If execution fails

to_dict()[source]

Convert tool to dictionary representation for serialization.

Return type:: Dict[str, Any]

class praval.core.tool_registry.ToolRegistry[source]

Bases: object

Centralized registry for managing tools and their relationships to agents.

The registry provides functionality to: - Register and retrieve tools - Associate tools with agents - Manage shared tools - Query tools by category - Handle tool lifecycle

__init__()[source]: Initialize the tool registry.

register_tool(tool)[source]

Register a tool in the registry.

Parameters:: tool (Tool) – Tool instance to register
Raises:: ToolError – If tool name already exists or registration fails
Return type:: None

get_tool(tool_name)[source]

Retrieve a tool by name.

Parameters:: tool_name (str) – Name of the tool to retrieve
Return type:: Optional[Tool]
Returns:: Tool instance if found, None otherwise

get_tools_for_agent(agent_name)[source]

Get all tools available to a specific agent.

This includes: - Tools owned by the agent - Shared tools - Tools explicitly assigned to the agent

Parameters:: agent_name (str) – Name of the agent
Return type:: List[Tool]
Returns:: List of Tool instances available to the agent

get_tools_by_category(category)[source]

Get all tools in a specific category.

Parameters:: category (str) – Category name
Return type:: List[Tool]
Returns:: List of Tool instances in the category

get_shared_tools()[source]

Get all shared tools.

Return type:: List[Tool]
Returns:: List of all shared Tool instances

list_all_tools()[source]

List all registered tools.

Return type:: List[Tool]
Returns:: List of all Tool instances

assign_tool_to_agent(tool_name, agent_name)[source]

Assign a tool to an agent at runtime.

Parameters:

tool_name (str) – Name of the tool to assign
agent_name (str) – Name of the agent to assign to

Return type:

Returns:

True if assignment successful, False if tool doesn’t exist

remove_tool_from_agent(tool_name, agent_name)[source]

Remove a tool assignment from an agent.

Parameters:

tool_name (str) – Name of the tool to remove
agent_name (str) – Name of the agent to remove from

Return type:

Returns:

True if removal successful, False if assignment didn’t exist

unregister_tool(tool_name)[source]

Unregister a tool from the registry.

Parameters:: tool_name (str) – Name of the tool to unregister
Return type:: bool
Returns:: True if unregistration successful, False if tool didn’t exist

clear_registry()[source]

Clear all tools from the registry.

Return type:: None

get_registry_stats()[source]

Get statistics about the registry.

Return type:: Dict[str, Any]
Returns:: Dictionary with registry statistics

search_tools(name_pattern=None, category=None, owned_by=None, shared_only=False, tags=None)[source]

Search for tools based on multiple criteria.

Parameters:

name_pattern (Optional[str]) – Pattern to match in tool names (case-insensitive)
category (Optional[str]) – Specific category to filter by
owned_by (Optional[str]) – Specific owner to filter by
shared_only (bool) – Only return shared tools
tags (Optional[List[str]]) – Tags that tools must have (any match)

Return type:

List[Tool]

Returns:

List of Tool instances matching the criteria

praval.core.tool_registry.get_tool_registry()[source]

Get the global tool registry instance.

Return type:: ToolRegistry
Returns:: Global ToolRegistry instance

praval.core.tool_registry.reset_tool_registry()[source]

Reset the global tool registry (primarily for testing).

Warning: This will clear all registered tools.

Return type:: None

Providers Module

Factory for creating LLM provider instances.

Provides a unified interface for instantiating different LLM providers with consistent configuration handling.

class praval.providers.factory.ProviderFactory[source]

Bases: object

Factory class for creating LLM provider instances.

static create_provider(provider_name, config)[source]

Create an LLM provider instance.

Parameters:

provider_name (str) – Name of the provider (openai, anthropic, cohere)
config (Any) – Configuration object for the provider

Returns:

Provider instance

Raises:

ProviderError – If provider is not supported or creation fails

OpenAI provider implementation for Praval framework.

Provides integration with OpenAI’s Chat Completions API with support for conversation history, tool calling, and streaming responses.

class praval.providers.openai.OpenAIProvider(config)[source]

Bases: object

OpenAI provider for LLM interactions.

Handles communication with OpenAI’s GPT models through the Chat Completions API with support for tools and conversation history.

__init__(config)[source]

Initialize OpenAI provider.

Parameters:: config – AgentConfig object with provider settings
Raises:: ProviderError – If OpenAI client initialization fails

generate(messages, tools=None)[source]

Generate a response using OpenAI’s Chat Completions API.

Parameters:

messages (List[Dict[str, str]]) – Conversation history as list of message dictionaries
tools (Optional[List[Dict[str, Any]]]) – Optional list of available tools for function calling

Return type:

Returns:

Generated response as a string

Raises:

ProviderError – If API call fails

Anthropic provider implementation for Praval framework.

Provides integration with Anthropic’s Claude models through the Messages API with support for conversation history and system messages.

class praval.providers.anthropic.AnthropicProvider(config)[source]

Bases: object

Anthropic provider for LLM interactions.

Handles communication with Anthropic’s Claude models through the Messages API with proper system message handling.

__init__(config)[source]

Initialize Anthropic provider.

Parameters:: config – AgentConfig object with provider settings
Raises:: ProviderError – If Anthropic client initialization fails

generate(messages, tools=None)[source]

Generate a response using Anthropic’s Messages API.

Parameters:

messages (List[Dict[str, str]]) – Conversation history as list of message dictionaries
tools (Optional[List[Dict[str, Any]]]) – Optional list of available tools (not fully supported yet)

Return type:

Returns:

Generated response as a string

Raises:

ProviderError – If API call fails

Cohere provider implementation for Praval framework.

Provides integration with Cohere’s chat models through their Chat API with support for conversation history.

class praval.providers.cohere.CohereProvider(config)[source]

Bases: object

Cohere provider for LLM interactions.

Handles communication with Cohere’s chat models through the Chat API with conversation history support.

__init__(config)[source]

Initialize Cohere provider.

Parameters:: config – AgentConfig object with provider settings
Raises:: ProviderError – If Cohere client initialization fails

generate(messages, tools=None)[source]

Generate a response using Cohere’s Chat API.

Parameters:

messages (List[Dict[str, str]]) – Conversation history as list of message dictionaries
tools (Optional[List[Dict[str, Any]]]) – Optional list of available tools (not fully supported yet)

Return type: