agentconnect.prompts.tools module

Tool definitions for the AgentConnect framework.

This module provides tools that agents can use to interact with each other, search for specialized agents, and collaborate on tasks. These tools are designed to be used with LangGraph workflows and LLM-based agents.

Key components:

  • Agent search tools: Find agents with specific capabilities

  • Collaboration tools: Send requests to other agents and manage responses

  • Task decomposition tools: Break complex tasks into manageable subtasks

  • Tool registry: Central registry for managing available tools

The tools in this module are designed to be used within the agent’s workflow to enable seamless agent-to-agent communication and collaboration.

class Subtask(**data)

Bases: BaseModel

A subtask to be completed by an agent.

Parameters:
id: str
title: str
description: str
status: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class TaskDecompositionResult(**data)

Bases: BaseModel

The result of task decomposition.

Parameters:
subtasks: List[Subtask]
original_task: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class AgentSearchInput(**data)

Bases: BaseModel

Input schema for agent search.

Parameters:

  • capability_name (str)

  • limit (int)

capability_name: str
limit: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class AgentSearchOutput(**data)

Bases: BaseModel

Output schema for agent search.

Parameters:
agent_ids: List[str]
capabilities: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class SendCollaborationRequestInput(**data)

Bases: BaseModel

Input schema for sending a collaboration request.

Parameters:

  • target_agent_id (str)

  • task_description (str)

  • timeout (int)

  • extra_data (Any)

target_agent_id: str
task_description: str
timeout: int
class Config

Bases: object

Config for the SendCollaborationRequestInput.

extra = 'allow'
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class SendCollaborationRequestOutput(**data)

Bases: BaseModel

Output schema for sending a collaboration request.

Parameters:

  • success (bool)

  • response (str | None)

success: bool
response: Optional[str]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class TaskDecompositionInput(**data)

Bases: BaseModel

Input schema for task decomposition.

Parameters:

  • task_description (str)

  • max_subtasks (int)

task_description: str
max_subtasks: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class TaskDecompositionOutput(**data)

Bases: BaseModel

Output schema for task decomposition.

Parameters:

subtasks (List[Dict[str, Any]])

subtasks: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ToolRegistry

Bases: object

Registry for all available tools that can be used by agents.

This class provides a centralized registry for managing the tools available to agents. It allows for registering, retrieving, and categorizing tools, enabling agents to access the right tools for specific tasks.

Tools can be organized by categories (e.g., ‘collaboration’, ‘task_management’) to make it easier for agents to discover relevant tools.

register_tool(tool)

Register a tool in the registry.

Parameters:

tool (StructuredTool) – The StructuredTool to register

Return type:

None

Note

If a tool with the same name already exists, it will be overwritten.

get_tool(name)

Get a tool by name.

Parameters:

name (str) – The name of the tool to retrieve

Return type:

Optional[StructuredTool]

Returns:

The tool if found, None otherwise

get_all_tools()

Get all registered tools.

Return type:

List[StructuredTool]

Returns:

A list of all tools in the registry

get_tools_by_category(category)

Get all tools in a specific category.

Parameters:

category (str) – The category to filter tools by

Return type:

List[StructuredTool]

Returns:

A list of tools in the specified category

class PromptTools(agent_registry, communication_hub, llm=None)

Bases: object

Class for creating and managing tools for agent prompts.

This class is responsible for creating, registering, and managing the tools that agents can use to perform actions such as searching for other agents, sending collaboration requests, and decomposing tasks.

Each agent has its own isolated set of tools through a dedicated ToolRegistry instance, ensuring that tools are properly configured for the specific agent using them.

Parameters:
agent_registry

Registry for accessing agent information

communication_hub

Hub for agent communication

llm

Optional language model for tools that require LLM capabilities

_current_agent_id

ID of the agent currently using these tools

_tool_registry

Registry for managing available tools

_available_capabilities

Cached list of available capabilities

_agent_specific_tools_registered

Flag indicating if agent-specific tools are registered

create_tool_from_function(func, name, description, args_schema, category=None, coroutine=None)

Create a tool from a function with proper async support.

This method creates a LangChain StructuredTool that can be used in agent workflows. It supports both synchronous and asynchronous implementations of the tool, allowing for efficient handling of I/O-bound operations.

The tool is automatically registered in the tool registry with the specified category, making it available for agent use.

Parameters:

  • func (Callable[..., Any]) – The synchronous function implementation

  • name (str) – Name of the tool (must be unique)

  • description (str) – Description of the tool that will be shown to the agent

  • args_schema (Type[TypeVar(T, bound= BaseModel)]) – Pydantic model for the tool’s arguments validation

  • category (Optional[str]) – Optional category for the tool (e.g., ‘collaboration’, ‘task_management’)

  • coroutine (Optional[Callable[..., Awaitable[Any]]]) – Optional async implementation of the function for better performance

Return type:

StructuredTool

Returns:

A StructuredTool that can be used in LangChain workflows

Note

If both sync and async implementations are provided, the async version will be used when the agent is running in an async context.

create_agent_search_tool()

Create a tool for searching agents by capability.

Return type:

StructuredTool

create_send_collaboration_request_tool()

Create a tool for sending collaboration requests to other agents.

Return type:

StructuredTool

create_task_decomposition_tool()

Create a tool for decomposing complex tasks into subtasks.

This tool helps agents break down complex tasks into smaller, more manageable subtasks. It’s useful for organizing and prioritizing work, especially for multi-step processes that would be difficult to tackle as a single unit.

The tool uses the LLM to analyze the task and create a structured decomposition with clear, actionable subtasks. Each subtask includes a unique ID, title, and description.

Return type:

StructuredTool

Returns:

A StructuredTool for task decomposition that can be used in agent workflows

Note

If the LLM is not available, the tool will fall back to a simple rule-based decomposition.

set_current_agent(agent_id)

Set the current agent ID for context in tools.

This method establishes the context for which agent is using the tools, which is essential for agent-specific tools like collaboration requests. It also triggers the registration of agent-specific tools if they haven’t been registered yet.

Parameters:

agent_id (str) – The ID of the agent currently using the tools

Return type:

None

Note

This method logs whenever the agent context changes to help with debugging and tracing agent interactions.

get_tools_for_workflow(categories=None, agent_id=None)

Get tools for a specific workflow based on categories.

This method retrieves the appropriate tools for a workflow, optionally filtered by category. It’s used by agent workflows to get the tools they need for specific tasks.

Parameters:

  • categories (Optional[List[str]]) – List of tool categories to include (e.g., ‘collaboration’, ‘task_management’)

  • agent_id (Optional[str]) – ID of the agent that will use these tools (for logging only, doesn’t change context)

Return type:

List[StructuredTool]

Returns:

List of StructuredTool instances configured for the agent

Note

If categories is None, all tools in the registry will be returned. The agent_id parameter is used only for logging and doesn’t change the current agent context.

On this page
Edit on GitHub