Skip to main content

Table of Contents

Module spoon_ai.agents

Module spoon_ai.agents.toolcall

ToolCallAgent Objects​

class ToolCallAgent(ReActAgent)

tool_choices​

type: ignore

mcp_tools_cache_ttl​

5 minutes TTL

run​

async def run(request: Optional[str] = None) -> str

Override run method to handle finish_reason termination specially.

step​

async def step() -> str

Override the step method to handle finish_reason termination properly.

Module spoon_ai.agents.react

Module spoon_ai.agents.mcp_client_mixin

MCPClientMixin Objects​

class MCPClientMixin()

get_session​

@asynccontextmanager
async def get_session()

Get a session with robust resource management and cleanup.

Features:

  • Automatic session reuse per task
  • Resource limits to prevent exhaustion
  • Proper cleanup on cancellation/failure
  • Periodic cleanup of stale sessions

list_mcp_tools​

async def list_mcp_tools()

Get the list of available tools from the MCP server

call_mcp_tool​

async def call_mcp_tool(tool_name: str, **kwargs)

Call a tool on the MCP server

send_mcp_message​

async def send_mcp_message(recipient: str,
                           message: Union[str, Dict[str, Any]],
                           topic: Optional[str] = None,
                           metadata: Optional[Dict[str, Any]] = None) -> bool

Send a message to the MCP system

Arguments:

  • recipient - Recipient ID
  • message - Message content (string or dictionary)
  • topic - Message topic
  • metadata - Additional metadata

Returns:

  • bool - Whether the message was sent successfully

cleanup​

async def cleanup()

Enhanced cleanup method with comprehensive resource cleanup.

get_session_stats​

def get_session_stats() -> Dict[str, Any]

Get session statistics for monitoring.

Module spoon_ai.agents.spoon_react_mcp

SpoonReactMCP Objects​

class SpoonReactMCP(SpoonReactAI)

list_mcp_tools​

async def list_mcp_tools()

Return MCP tools from available_tools manager

Module spoon_ai.agents.monitor

Module spoon_ai.agents.rag

RetrievalMixin Objects​

class RetrievalMixin()

Mixin class for retrieval-augmented generation functionality

initialize_retrieval_client​

def initialize_retrieval_client(backend: str = 'chroma', **kwargs)

Initialize the retrieval client if it doesn't exist

add_documents​

def add_documents(documents, backend: str = 'chroma', **kwargs)

Add documents to the retrieval system

retrieve_relevant_documents​

def retrieve_relevant_documents(query, k=5, backend: str = 'chroma', **kwargs)

Retrieve relevant documents for a query

get_context_from_query​

def get_context_from_query(query)

Get context string from relevant documents for a query

Module spoon_ai.agents.custom_agent

CustomAgent Objects​

class CustomAgent(ToolCallAgent)

Custom Agent class allowing users to create their own agents and add custom tools

Usage: Create custom agent and add tools: agent = CustomAgent(name="my_agent", description="My custom agent") agent.add_tool(MyCustomTool()) result = await agent.run("Use my custom tool")

add_tool​

def add_tool(tool: BaseTool) -> None

Add a tool to the agent with validation.

Arguments:

  • tool - Tool instance to add

Raises:

  • ValueError - If tool is invalid or already exists

add_tools​

def add_tools(tools: List[BaseTool]) -> None

Add multiple tools to the agent with atomic operation.

Arguments:

  • tools - List of tool instances to add

Raises:

  • ValueError - If any tool is invalid

remove_tool​

def remove_tool(tool_name: str) -> bool

Remove a tool from the agent.

Arguments:

  • tool_name - Name of the tool to remove

Returns:

  • bool - True if tool was removed, False if not found

list_tools​

def list_tools() -> List[str]

List all available tools in the agent.

Returns:

List of tool names, empty list if no tools

get_tool_info​

def get_tool_info() -> Dict[str, Dict[str, Any]]

Get detailed information about all tools.

Returns:

Dictionary with tool names as keys and tool info as values

validate_tools​

def validate_tools() -> Dict[str, Any]

Validate all current tools and return validation report.

Returns:

Dictionary with validation results

run​

async def run(request: Optional[str] = None) -> str

Run the agent with enhanced tool validation.

Arguments:

  • request - User request

Returns:

Processing result

clear​

def clear()

Enhanced clear method with proper tool state management.

Module spoon_ai.agents.graph_agent

Graph-based agent implementation for SpoonOS.

This module provides the GraphAgent class that executes StateGraph workflows, integrating the graph execution system with the existing agent architecture.

GraphAgent Objects​

class GraphAgent(BaseAgent)

An agent that executes StateGraph workflows.

This agent provides a bridge between the existing SpoonOS agent architecture and the new graph-based execution system. It allows complex, stateful workflows to be defined as graphs and executed with proper state management.

Key Features:

  • Executes StateGraph workflows
  • Maintains compatibility with existing agent interfaces
  • Provides detailed execution logging and error handling
  • Supports both sync and async node functions

__init__​

def __init__(**kwargs)

Initialize the GraphAgent.

Arguments:

  • graph - StateGraph instance to execute
  • **kwargs - Additional arguments passed to BaseAgent

Raises:

  • ValueError - If no graph is provided

validate_graph​

@validator('graph')
def validate_graph(cls, v)

Validate that the provided graph is a StateGraph instance.

run​

async def run(request: Optional[str] = None) -> str

Execute the graph workflow.

This method overrides the base run method to invoke the compiled graph instead of the traditional step-based execution loop.

Arguments:

  • request - Optional input request to include in initial state

Returns:

String representation of the execution result

Raises:

  • RuntimeError - If agent is not in IDLE state
  • GraphExecutionError - If graph execution fails

step​

async def step() -> str

Step method for compatibility with BaseAgent.

Since GraphAgent uses graph execution instead of step-based execution, this method is not used in normal operation but is required by the BaseAgent interface.

Returns:

Status message indicating graph-based execution

get_execution_history​

def get_execution_history() -> list

Get the execution history from the last graph run.

Returns:

List of execution steps with metadata

get_execution_metadata​

def get_execution_metadata() -> Dict[str, Any]

Get metadata from the last execution.

Returns:

Dictionary containing execution metadata

clear_state​

def clear_state()

Clear preserved state and execution history.

update_initial_state​

def update_initial_state(updates: Dict[str, Any])

Update the initial state for future executions.

Arguments:

  • updates - Dictionary of state updates to merge

set_preserve_state​

def set_preserve_state(preserve: bool)

Enable or disable state preservation between runs.

Arguments:

  • preserve - Whether to preserve state between runs

Module spoon_ai.agents.spoon_react

create_configured_chatbot​

def create_configured_chatbot()

Create a ChatBot instance with intelligent provider selection.

SpoonReactAI Objects​

class SpoonReactAI(ToolCallAgent)

__init__​

def __init__(**kwargs)

Initialize SpoonReactAI with both ToolCallAgent and MCPClientMixin initialization

initialize​

async def initialize(__context: Any = None)

Initialize async components and subscribe to topics

run​

async def run(request: Optional[str] = None) -> str

Ensure prompts reflect current tools before running.

Module spoon_ai.agents.base

ThreadSafeOutputQueue Objects​

class ThreadSafeOutputQueue()

Thread-safe output queue with fair access and timeout protection

get​

async def get(timeout: Optional[float] = 30.0) -> Any

Get item with timeout and fair access

BaseAgent Objects​

class BaseAgent(BaseModel, ABC)

Thread-safe base class for all agents with proper concurrency handling.

add_message​

async def add_message(role: Literal["user", "assistant", "tool"],
                      content: str,
                      tool_call_id: Optional[str] = None,
                      tool_calls: Optional[List[ToolCall]] = None,
                      tool_name: Optional[str] = None,
                      timeout: Optional[float] = None) -> None

Thread-safe message addition with timeout protection

state_context​

@asynccontextmanager
async def state_context(new_state: AgentState,
                        timeout: Optional[float] = None)

Thread-safe state context manager with deadlock prevention. Acquires the state lock only to perform quick transitions, not for the duration of the work inside the context, avoiding long-held locks and false timeouts during network calls.

run​

async def run(request: Optional[str] = None,
              timeout: Optional[float] = None) -> str

Thread-safe run method with proper concurrency control and callback support.

step​

async def step(run_id: Optional[uuid.UUID] = None) -> str

Override this method in subclasses - now with step-level locking and callback support.

is_stuck​

async def is_stuck() -> bool

Thread-safe stuck detection

handle_stuck_state​

async def handle_stuck_state()

Thread-safe stuck state handling

add_documents​

def add_documents(documents) -> None

Store documents on the agent so CLI load-docs works without RAG mixin.

This default implementation keeps the documents in-memory under self._loaded_documents. Agents that support retrieval should override this method to index documents into their vector store.

save_chat_history​

def save_chat_history()

Thread-safe chat history saving

stream​

async def stream(timeout: Optional[float] = None)

Thread-safe streaming with proper cleanup and timeout

process_mcp_message​

async def process_mcp_message(content: Any,
                              sender: str,
                              message: Dict[str, Any],
                              agent_id: str,
                              timeout: Optional[float] = None)

Thread-safe MCP message processing with timeout protection

shutdown​

async def shutdown(timeout: float = 30.0)

Graceful shutdown with cleanup of active operations

get_diagnostics​

def get_diagnostics() -> Dict[str, Any]

Get diagnostic information about the agent's state