Tool node

ToolNode package.

This package provides a modularized implementation of ToolNode. Public API:

• ToolNode
• HAS_FASTMCP, HAS_MCP

Backwards-compatible import path: from pyagenity.graph.tool_node import ToolNode

Modules:

Name	Description
base	Tool execution node for PyAgenity graph workflows.
constants	Constants for ToolNode package.
deps	Dependency flags and optional imports for ToolNode.
executors	Executors for different tool providers and local functions.
schema	Schema utilities and local tool description building for ToolNode.

Classes:

Name	Description
ToolNode	A unified registry and executor for callable functions from various tool providers.

Attributes:

Name	Type	Description
HAS_FASTMCP
HAS_MCP

Attributes

HAS_FASTMCP module-attribute

HAS_FASTMCP = True

HAS_MCP module-attribute

HAS_MCP = True

__all__ module-attribute

__all__ = ['HAS_FASTMCP', 'HAS_MCP', 'ToolNode']

Classes

ToolNode

Bases: SchemaMixin, LocalExecMixin, MCPMixin, ComposioMixin, LangChainMixin, KwargsResolverMixin

A unified registry and executor for callable functions from various tool providers.

ToolNode serves as the central hub for managing and executing tools from multiple sources: - Local Python functions - MCP (Model Context Protocol) tools - Composio adapter tools - LangChain tools

The class uses a mixin-based architecture to separate concerns and maintain clean integration with different tool providers. It provides both synchronous and asynchronous execution methods with comprehensive event publishing and error handling.

Attributes:

Name	Type	Description
_funcs	dict[str, Callable]	Dictionary mapping function names to callable functions.
_client	Client | None	Optional MCP client for remote tool execution.
_composio	ComposioAdapter | None	Optional Composio adapter for external integrations.
_langchain	Any | None	Optional LangChain adapter for LangChain tools.
mcp_tools	list[str]	List of available MCP tool names.
composio_tools	list[str]	List of available Composio tool names.
langchain_tools	list[str]	List of available LangChain tool names.

Example

# Define local tools
def weather_tool(location: str) -> str:
    return f"Weather in {location}: Sunny, 25°C"


def calculator(a: int, b: int) -> int:
    return a + b


# Create ToolNode with local functions
tools = ToolNode([weather_tool, calculator])

# Execute a tool
result = await tools.invoke(
    name="weather_tool",
    args={"location": "New York"},
    tool_call_id="call_123",
    config={"user_id": "user1"},
    state=agent_state,
)

Methods:

Name	Description
__init__	Initialize ToolNode with functions and optional tool adapters.
all_tools	Get all available tools from all configured providers.
all_tools_sync	Synchronously get all available tools from all configured providers.
get_local_tool	Generate OpenAI-compatible tool definitions for all registered local functions.
invoke	Execute a specific tool by name with the provided arguments.
stream	Execute a tool with streaming support, yielding incremental results.

Source code in pyagenity/graph/tool_node/base.py

class ToolNode(
    SchemaMixin,
    LocalExecMixin,
    MCPMixin,
    ComposioMixin,
    LangChainMixin,
    KwargsResolverMixin,
):
    """A unified registry and executor for callable functions from various tool providers.

    ToolNode serves as the central hub for managing and executing tools from multiple sources:
    - Local Python functions
    - MCP (Model Context Protocol) tools
    - Composio adapter tools
    - LangChain tools

    The class uses a mixin-based architecture to separate concerns and maintain clean
    integration with different tool providers. It provides both synchronous and asynchronous
    execution methods with comprehensive event publishing and error handling.

    Attributes:
        _funcs: Dictionary mapping function names to callable functions.
        _client: Optional MCP client for remote tool execution.
        _composio: Optional Composio adapter for external integrations.
        _langchain: Optional LangChain adapter for LangChain tools.
        mcp_tools: List of available MCP tool names.
        composio_tools: List of available Composio tool names.
        langchain_tools: List of available LangChain tool names.

    Example:
        ```python
        # Define local tools
        def weather_tool(location: str) -> str:
            return f"Weather in {location}: Sunny, 25°C"


        def calculator(a: int, b: int) -> int:
            return a + b


        # Create ToolNode with local functions
        tools = ToolNode([weather_tool, calculator])

        # Execute a tool
        result = await tools.invoke(
            name="weather_tool",
            args={"location": "New York"},
            tool_call_id="call_123",
            config={"user_id": "user1"},
            state=agent_state,
        )
        ```
    """

    def __init__(
        self,
        functions: t.Iterable[t.Callable],
        client: deps.Client | None = None,  # type: ignore
        composio_adapter: ComposioAdapter | None = None,
        langchain_adapter: t.Any | None = None,
    ) -> None:
        """Initialize ToolNode with functions and optional tool adapters.

        Args:
            functions: Iterable of callable functions to register as tools. Each function
                will be registered with its `__name__` as the tool identifier.
            client: Optional MCP (Model Context Protocol) client for remote tool access.
                Requires 'fastmcp' and 'mcp' packages to be installed.
            composio_adapter: Optional Composio adapter for external integrations and
                third-party API access.
            langchain_adapter: Optional LangChain adapter for accessing LangChain tools
                and integrations.

        Raises:
            ImportError: If MCP client is provided but required packages are not installed.
            TypeError: If any item in functions is not callable.

        Note:
            When using MCP client functionality, ensure you have installed the required
            dependencies with: `pip install pyagenity[mcp]`
        """
        logger.info("Initializing ToolNode with %d functions", len(list(functions)))

        if client is not None:
            # Read flags dynamically so tests can patch pyagenity.graph.tool_node.HAS_*
            mod = sys.modules.get("pyagenity.graph.tool_node")
            has_fastmcp = getattr(mod, "HAS_FASTMCP", deps.HAS_FASTMCP) if mod else deps.HAS_FASTMCP
            has_mcp = getattr(mod, "HAS_MCP", deps.HAS_MCP) if mod else deps.HAS_MCP

            if not has_fastmcp or not has_mcp:
                raise ImportError(
                    "MCP client functionality requires 'fastmcp' and 'mcp' packages. "
                    "Install with: pip install pyagenity[mcp]"
                )
            logger.debug("ToolNode initialized with MCP client")

        self._funcs: dict[str, t.Callable] = {}
        self._client: deps.Client | None = client  # type: ignore
        self._composio: ComposioAdapter | None = composio_adapter
        self._langchain: t.Any | None = langchain_adapter

        for fn in functions:
            if not callable(fn):
                raise TypeError("ToolNode only accepts callables")
            self._funcs[fn.__name__] = fn

        self.mcp_tools: list[str] = []
        self.composio_tools: list[str] = []
        self.langchain_tools: list[str] = []

    async def _all_tools_async(self) -> list[dict]:
        tools: list[dict] = self.get_local_tool()
        tools.extend(await self._get_mcp_tool())
        tools.extend(await self._get_composio_tools())
        tools.extend(await self._get_langchain_tools())
        return tools

    async def all_tools(self) -> list[dict]:
        """Get all available tools from all configured providers.

        Retrieves and combines tool definitions from local functions, MCP client,
        Composio adapter, and LangChain adapter. Each tool definition includes
        the function schema with parameters and descriptions.

        Returns:
            List of tool definitions in OpenAI function calling format. Each dict
            contains 'type': 'function' and 'function' with name, description,
            and parameters schema.

        Example:
            ```python
            tools = await tool_node.all_tools()
            # Returns:
            # [
            #   {
            #     "type": "function",
            #     "function": {
            #       "name": "weather_tool",
            #       "description": "Get weather information for a location",
            #       "parameters": {
            #         "type": "object",
            #         "properties": {
            #           "location": {"type": "string"}
            #         },
            #         "required": ["location"]
            #       }
            #     }
            #   }
            # ]
            ```
        """
        return await self._all_tools_async()

    def all_tools_sync(self) -> list[dict]:
        """Synchronously get all available tools from all configured providers.

        This is a synchronous wrapper around the async all_tools() method.
        It uses asyncio.run() to handle async operations from MCP, Composio,
        and LangChain adapters.

        Returns:
            List of tool definitions in OpenAI function calling format.

        Note:
            Prefer using the async `all_tools()` method when possible, especially
            in async contexts, to avoid potential event loop issues.
        """
        tools: list[dict] = self.get_local_tool()
        if self._client:
            result = asyncio.run(self._get_mcp_tool())
            if result:
                tools.extend(result)
        comp = asyncio.run(self._get_composio_tools())
        if comp:
            tools.extend(comp)
        lc = asyncio.run(self._get_langchain_tools())
        if lc:
            tools.extend(lc)
        return tools

    async def invoke(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        state: AgentState,
        callback_manager: CallbackManager = Inject[CallbackManager],
    ) -> t.Any:
        """Execute a specific tool by name with the provided arguments.

        This method handles tool execution across all configured providers (local,
        MCP, Composio, LangChain) with comprehensive error handling, event publishing,
        and callback management.

        Args:
            name: The name of the tool to execute.
            args: Dictionary of arguments to pass to the tool function.
            tool_call_id: Unique identifier for this tool execution, used for
                tracking and result correlation.
            config: Configuration dictionary containing execution context and
                user-specific settings.
            state: Current agent state for context-aware tool execution.
            callback_manager: Manager for executing pre/post execution callbacks.
                Injected via dependency injection if not provided.

        Returns:
            Message object containing tool execution results, either successful
            output or error information with appropriate status indicators.

        Raises:
            The method handles all exceptions internally and returns error Messages
            rather than raising exceptions, ensuring robust execution flow.

        Example:
            ```python
            result = await tool_node.invoke(
                name="weather_tool",
                args={"location": "Paris", "units": "metric"},
                tool_call_id="call_abc123",
                config={"user_id": "user1", "session_id": "session1"},
                state=current_agent_state,
            )

            # result is a Message with tool execution results
            print(result.content)  # Tool output or error information
            ```

        Note:
            The method publishes execution events throughout the process for
            monitoring and debugging purposes. Tool execution is routed based
            on tool provider precedence: MCP → Composio → LangChain → Local.
        """
        logger.info("Executing tool '%s' with %d arguments", name, len(args))
        logger.debug("Tool arguments: %s", args)

        event = EventModel.default(
            config,
            data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.node_name = name
        # Attach structured tool call block
        with contextlib.suppress(Exception):
            event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]
        publish_event(event)

        if name in self.mcp_tools:
            event.metadata["is_mcp"] = True
            publish_event(event)
            res = await self._mcp_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            # Attach tool result block mirroring the tool output
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self.composio_tools:
            event.metadata["is_composio"] = True
            publish_event(event)
            res = await self._composio_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self.langchain_tools:
            event.metadata["is_langchain"] = True
            publish_event(event)
            res = await self._langchain_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self._funcs:
            event.metadata["is_mcp"] = False
            publish_event(event)
            res = await self._internal_execute(
                name,
                args,
                tool_call_id,
                config,
                state,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        error_msg = f"Tool '{name}' not found."
        event.data["error"] = error_msg
        event.event_type = EventType.ERROR
        event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
        publish_event(event)
        return Message.tool_message(
            content=[
                ErrorBlock(message=error_msg),
                ToolResultBlock(
                    call_id=tool_call_id,
                    output=error_msg,
                    is_error=True,
                    status="failed",
                ),
            ],
        )

    async def stream(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        state: AgentState,
        callback_manager: CallbackManager = Inject[CallbackManager],
    ) -> t.AsyncIterator[Message]:
        """Execute a tool with streaming support, yielding incremental results.

        Similar to invoke() but designed for tools that can provide streaming responses
        or when you want to process results as they become available. Currently,
        most tool providers return complete results, so this method typically yields
        a single Message with the full result.

        Args:
            name: The name of the tool to execute.
            args: Dictionary of arguments to pass to the tool function.
            tool_call_id: Unique identifier for this tool execution.
            config: Configuration dictionary containing execution context.
            state: Current agent state for context-aware tool execution.
            callback_manager: Manager for executing pre/post execution callbacks.

        Yields:
            Message objects containing tool execution results or status updates.
            For most tools, this will yield a single complete result Message.

        Example:
            ```python
            async for message in tool_node.stream(
                name="data_processor",
                args={"dataset": "large_data.csv"},
                tool_call_id="call_stream123",
                config={"user_id": "user1"},
                state=current_state,
            ):
                print(f"Received: {message.content}")
                # Process each streamed result
            ```

        Note:
            The streaming interface is designed for future expansion where tools
            may provide true streaming responses. Currently, it provides a
            consistent async iterator interface over tool results.
        """
        logger.info("Executing tool '%s' with %d arguments", name, len(args))
        logger.debug("Tool arguments: %s", args)
        event = EventModel.default(
            config,
            data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.node_name = "ToolNode"
        with contextlib.suppress(Exception):
            event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]

        if name in self.mcp_tools:
            event.metadata["function_type"] = "mcp"
            publish_event(event)
            message = await self._mcp_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self.composio_tools:
            event.metadata["function_type"] = "composio"
            publish_event(event)
            message = await self._composio_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self.langchain_tools:
            event.metadata["function_type"] = "langchain"
            publish_event(event)
            message = await self._langchain_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self._funcs:
            event.metadata["function_type"] = "internal"
            publish_event(event)

            result = await self._internal_execute(
                name,
                args,
                tool_call_id,
                config,
                state,
                callback_manager,
            )
            event.data["message"] = result.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=result.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield result
            return

        error_msg = f"Tool '{name}' not found."
        event.data["error"] = error_msg
        event.event_type = EventType.ERROR
        event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
        publish_event(event)

        yield Message.tool_message(
            content=[
                ErrorBlock(message=error_msg),
                ToolResultBlock(
                    call_id=tool_call_id,
                    output=error_msg,
                    is_error=True,
                    status="failed",
                ),
            ],
        )

Attributes

composio_tools instance-attribute

composio_tools = []

langchain_tools instance-attribute

langchain_tools = []

mcp_tools instance-attribute

mcp_tools = []

Functions

__init__

__init__(functions, client=None, composio_adapter=None, langchain_adapter=None)

Initialize ToolNode with functions and optional tool adapters.

Parameters:

Name	Type	Description	Default
functions	Iterable[Callable]	Iterable of callable functions to register as tools. Each function will be registered with its __name__ as the tool identifier.	required
client	Client | None	Optional MCP (Model Context Protocol) client for remote tool access. Requires 'fastmcp' and 'mcp' packages to be installed.	None
composio_adapter	ComposioAdapter | None	Optional Composio adapter for external integrations and third-party API access.	None
langchain_adapter	Any | None	Optional LangChain adapter for accessing LangChain tools and integrations.	None

Raises:

Type	Description
ImportError	If MCP client is provided but required packages are not installed.
TypeError	If any item in functions is not callable.

Note

When using MCP client functionality, ensure you have installed the required dependencies with: pip install pyagenity[mcp]

Source code in pyagenity/graph/tool_node/base.py

def __init__(
    self,
    functions: t.Iterable[t.Callable],
    client: deps.Client | None = None,  # type: ignore
    composio_adapter: ComposioAdapter | None = None,
    langchain_adapter: t.Any | None = None,
) -> None:
    """Initialize ToolNode with functions and optional tool adapters.

    Args:
        functions: Iterable of callable functions to register as tools. Each function
            will be registered with its `__name__` as the tool identifier.
        client: Optional MCP (Model Context Protocol) client for remote tool access.
            Requires 'fastmcp' and 'mcp' packages to be installed.
        composio_adapter: Optional Composio adapter for external integrations and
            third-party API access.
        langchain_adapter: Optional LangChain adapter for accessing LangChain tools
            and integrations.

    Raises:
        ImportError: If MCP client is provided but required packages are not installed.
        TypeError: If any item in functions is not callable.

    Note:
        When using MCP client functionality, ensure you have installed the required
        dependencies with: `pip install pyagenity[mcp]`
    """
    logger.info("Initializing ToolNode with %d functions", len(list(functions)))

    if client is not None:
        # Read flags dynamically so tests can patch pyagenity.graph.tool_node.HAS_*
        mod = sys.modules.get("pyagenity.graph.tool_node")
        has_fastmcp = getattr(mod, "HAS_FASTMCP", deps.HAS_FASTMCP) if mod else deps.HAS_FASTMCP
        has_mcp = getattr(mod, "HAS_MCP", deps.HAS_MCP) if mod else deps.HAS_MCP

        if not has_fastmcp or not has_mcp:
            raise ImportError(
                "MCP client functionality requires 'fastmcp' and 'mcp' packages. "
                "Install with: pip install pyagenity[mcp]"
            )
        logger.debug("ToolNode initialized with MCP client")

    self._funcs: dict[str, t.Callable] = {}
    self._client: deps.Client | None = client  # type: ignore
    self._composio: ComposioAdapter | None = composio_adapter
    self._langchain: t.Any | None = langchain_adapter

    for fn in functions:
        if not callable(fn):
            raise TypeError("ToolNode only accepts callables")
        self._funcs[fn.__name__] = fn

    self.mcp_tools: list[str] = []
    self.composio_tools: list[str] = []
    self.langchain_tools: list[str] = []

all_tools async

all_tools()

Get all available tools from all configured providers.

Retrieves and combines tool definitions from local functions, MCP client, Composio adapter, and LangChain adapter. Each tool definition includes the function schema with parameters and descriptions.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format. Each dict
list[dict]	contains 'type': 'function' and 'function' with name, description,
list[dict]	and parameters schema.

Example

tools = await tool_node.all_tools()
# Returns:
# [
#   {
#     "type": "function",
#     "function": {
#       "name": "weather_tool",
#       "description": "Get weather information for a location",
#       "parameters": {
#         "type": "object",
#         "properties": {
#           "location": {"type": "string"}
#         },
#         "required": ["location"]
#       }
#     }
#   }
# ]

Source code in pyagenity/graph/tool_node/base.py

async def all_tools(self) -> list[dict]:
    """Get all available tools from all configured providers.

    Retrieves and combines tool definitions from local functions, MCP client,
    Composio adapter, and LangChain adapter. Each tool definition includes
    the function schema with parameters and descriptions.

    Returns:
        List of tool definitions in OpenAI function calling format. Each dict
        contains 'type': 'function' and 'function' with name, description,
        and parameters schema.

    Example:
        ```python
        tools = await tool_node.all_tools()
        # Returns:
        # [
        #   {
        #     "type": "function",
        #     "function": {
        #       "name": "weather_tool",
        #       "description": "Get weather information for a location",
        #       "parameters": {
        #         "type": "object",
        #         "properties": {
        #           "location": {"type": "string"}
        #         },
        #         "required": ["location"]
        #       }
        #     }
        #   }
        # ]
        ```
    """
    return await self._all_tools_async()

all_tools_sync

all_tools_sync()

Synchronously get all available tools from all configured providers.

This is a synchronous wrapper around the async all_tools() method. It uses asyncio.run() to handle async operations from MCP, Composio, and LangChain adapters.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format.

Note

Prefer using the async all_tools() method when possible, especially in async contexts, to avoid potential event loop issues.

Source code in pyagenity/graph/tool_node/base.py

def all_tools_sync(self) -> list[dict]:
    """Synchronously get all available tools from all configured providers.

    This is a synchronous wrapper around the async all_tools() method.
    It uses asyncio.run() to handle async operations from MCP, Composio,
    and LangChain adapters.

    Returns:
        List of tool definitions in OpenAI function calling format.

    Note:
        Prefer using the async `all_tools()` method when possible, especially
        in async contexts, to avoid potential event loop issues.
    """
    tools: list[dict] = self.get_local_tool()
    if self._client:
        result = asyncio.run(self._get_mcp_tool())
        if result:
            tools.extend(result)
    comp = asyncio.run(self._get_composio_tools())
    if comp:
        tools.extend(comp)
    lc = asyncio.run(self._get_langchain_tools())
    if lc:
        tools.extend(lc)
    return tools

get_local_tool

get_local_tool()

Generate OpenAI-compatible tool definitions for all registered local functions.

Inspects all registered functions in _funcs and automatically generates tool schemas by analyzing function signatures, type annotations, and docstrings. Excludes injectable parameters that are provided by the framework.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format. Each
list[dict]	definition includes the function name, description (from docstring),
list[dict]	and complete parameter schema with types and required fields.

Example

For a function:

def calculate(a: int, b: int, operation: str = "add") -> int:
    '''Perform arithmetic calculation.'''
    return a + b if operation == "add" else a - b

Returns:

[
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Perform arithmetic calculation.",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "integer"},
                    "b": {"type": "integer"},
                    "operation": {"type": "string", "default": "add"},
                },
                "required": ["a", "b"],
            },
        },
    }
]

Note

Parameters listed in INJECTABLE_PARAMS (like 'state', 'config', 'tool_call_id') are automatically excluded from the generated schema as they are provided by the framework during execution.

Source code in pyagenity/graph/tool_node/schema.py

def get_local_tool(self) -> list[dict]:
    """Generate OpenAI-compatible tool definitions for all registered local functions.

    Inspects all registered functions in _funcs and automatically generates
    tool schemas by analyzing function signatures, type annotations, and docstrings.
    Excludes injectable parameters that are provided by the framework.

    Returns:
        List of tool definitions in OpenAI function calling format. Each
        definition includes the function name, description (from docstring),
        and complete parameter schema with types and required fields.

    Example:
        For a function:
        ```python
        def calculate(a: int, b: int, operation: str = "add") -> int:
            '''Perform arithmetic calculation.'''
            return a + b if operation == "add" else a - b
        ```

        Returns:
        ```python
        [
            {
                "type": "function",
                "function": {
                    "name": "calculate",
                    "description": "Perform arithmetic calculation.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "a": {"type": "integer"},
                            "b": {"type": "integer"},
                            "operation": {"type": "string", "default": "add"},
                        },
                        "required": ["a", "b"],
                    },
                },
            }
        ]
        ```

    Note:
        Parameters listed in INJECTABLE_PARAMS (like 'state', 'config',
        'tool_call_id') are automatically excluded from the generated schema
        as they are provided by the framework during execution.
    """
    tools: list[dict] = []
    for name, fn in self._funcs.items():
        sig = inspect.signature(fn)
        params_schema: dict = {"type": "object", "properties": {}, "required": []}

        for p_name, p in sig.parameters.items():
            if p.kind in (
                inspect.Parameter.VAR_POSITIONAL,
                inspect.Parameter.VAR_KEYWORD,
            ):
                continue

            if p_name in INJECTABLE_PARAMS:
                continue

            annotation = p.annotation if p.annotation is not inspect._empty else str
            prop = SchemaMixin._annotation_to_schema(annotation, p.default)
            params_schema["properties"][p_name] = prop

            if p.default is inspect._empty:
                params_schema["required"].append(p_name)

        if not params_schema["required"]:
            params_schema.pop("required")

        description = inspect.getdoc(fn) or "No description provided."

        # provider = getattr(fn, "_py_tool_provider", None)
        # tags = getattr(fn, "_py_tool_tags", None)
        # capabilities = getattr(fn, "_py_tool_capabilities", None)

        entry = {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": params_schema,
            },
        }
        # meta: dict[str, t.Any] = {}
        # if provider:
        #     meta["provider"] = provider
        # if tags:
        #     meta["tags"] = tags
        # if capabilities:
        #     meta["capabilities"] = capabilities
        # if meta:
        #     entry["x-pyagenity"] = meta

        tools.append(entry)

    return tools

invoke async

invoke(name, args, tool_call_id, config, state, callback_manager=Inject[CallbackManager])

Execute a specific tool by name with the provided arguments.

This method handles tool execution across all configured providers (local, MCP, Composio, LangChain) with comprehensive error handling, event publishing, and callback management.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to execute.	required
args	dict	Dictionary of arguments to pass to the tool function.	required
tool_call_id	str	Unique identifier for this tool execution, used for tracking and result correlation.	required
config	dict[str, Any]	Configuration dictionary containing execution context and user-specific settings.	required
state	AgentState	Current agent state for context-aware tool execution.	required
callback_manager	CallbackManager	Manager for executing pre/post execution callbacks. Injected via dependency injection if not provided.	Inject[CallbackManager]

Returns:

Type	Description
Any	Message object containing tool execution results, either successful
Any	output or error information with appropriate status indicators.

Example

result = await tool_node.invoke(
    name="weather_tool",
    args={"location": "Paris", "units": "metric"},
    tool_call_id="call_abc123",
    config={"user_id": "user1", "session_id": "session1"},
    state=current_agent_state,
)

# result is a Message with tool execution results
print(result.content)  # Tool output or error information

Note

The method publishes execution events throughout the process for monitoring and debugging purposes. Tool execution is routed based on tool provider precedence: MCP → Composio → LangChain → Local.

Source code in pyagenity/graph/tool_node/base.py

async def invoke(  # noqa: PLR0915
    self,
    name: str,
    args: dict,
    tool_call_id: str,
    config: dict[str, t.Any],
    state: AgentState,
    callback_manager: CallbackManager = Inject[CallbackManager],
) -> t.Any:
    """Execute a specific tool by name with the provided arguments.

    This method handles tool execution across all configured providers (local,
    MCP, Composio, LangChain) with comprehensive error handling, event publishing,
    and callback management.

    Args:
        name: The name of the tool to execute.
        args: Dictionary of arguments to pass to the tool function.
        tool_call_id: Unique identifier for this tool execution, used for
            tracking and result correlation.
        config: Configuration dictionary containing execution context and
            user-specific settings.
        state: Current agent state for context-aware tool execution.
        callback_manager: Manager for executing pre/post execution callbacks.
            Injected via dependency injection if not provided.

    Returns:
        Message object containing tool execution results, either successful
        output or error information with appropriate status indicators.

    Raises:
        The method handles all exceptions internally and returns error Messages
        rather than raising exceptions, ensuring robust execution flow.

    Example:
        ```python
        result = await tool_node.invoke(
            name="weather_tool",
            args={"location": "Paris", "units": "metric"},
            tool_call_id="call_abc123",
            config={"user_id": "user1", "session_id": "session1"},
            state=current_agent_state,
        )

        # result is a Message with tool execution results
        print(result.content)  # Tool output or error information
        ```

    Note:
        The method publishes execution events throughout the process for
        monitoring and debugging purposes. Tool execution is routed based
        on tool provider precedence: MCP → Composio → LangChain → Local.
    """
    logger.info("Executing tool '%s' with %d arguments", name, len(args))
    logger.debug("Tool arguments: %s", args)

    event = EventModel.default(
        config,
        data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
        content_type=[ContentType.TOOL_CALL],
        event=Event.TOOL_EXECUTION,
    )
    event.node_name = name
    # Attach structured tool call block
    with contextlib.suppress(Exception):
        event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]
    publish_event(event)

    if name in self.mcp_tools:
        event.metadata["is_mcp"] = True
        publish_event(event)
        res = await self._mcp_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        # Attach tool result block mirroring the tool output
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self.composio_tools:
        event.metadata["is_composio"] = True
        publish_event(event)
        res = await self._composio_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self.langchain_tools:
        event.metadata["is_langchain"] = True
        publish_event(event)
        res = await self._langchain_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self._funcs:
        event.metadata["is_mcp"] = False
        publish_event(event)
        res = await self._internal_execute(
            name,
            args,
            tool_call_id,
            config,
            state,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    error_msg = f"Tool '{name}' not found."
    event.data["error"] = error_msg
    event.event_type = EventType.ERROR
    event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
    publish_event(event)
    return Message.tool_message(
        content=[
            ErrorBlock(message=error_msg),
            ToolResultBlock(
                call_id=tool_call_id,
                output=error_msg,
                is_error=True,
                status="failed",
            ),
        ],
    )

stream async

stream(name, args, tool_call_id, config, state, callback_manager=Inject[CallbackManager])

Execute a tool with streaming support, yielding incremental results.

Similar to invoke() but designed for tools that can provide streaming responses or when you want to process results as they become available. Currently, most tool providers return complete results, so this method typically yields a single Message with the full result.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to execute.	required
args	dict	Dictionary of arguments to pass to the tool function.	required
tool_call_id	str	Unique identifier for this tool execution.	required
config	dict[str, Any]	Configuration dictionary containing execution context.	required
state	AgentState	Current agent state for context-aware tool execution.	required
callback_manager	CallbackManager	Manager for executing pre/post execution callbacks.	Inject[CallbackManager]

Yields:

Type	Description
AsyncIterator[Message]	Message objects containing tool execution results or status updates.
AsyncIterator[Message]	For most tools, this will yield a single complete result Message.

Example

async for message in tool_node.stream(
    name="data_processor",
    args={"dataset": "large_data.csv"},
    tool_call_id="call_stream123",
    config={"user_id": "user1"},
    state=current_state,
):
    print(f"Received: {message.content}")
    # Process each streamed result

Note

The streaming interface is designed for future expansion where tools may provide true streaming responses. Currently, it provides a consistent async iterator interface over tool results.

Source code in pyagenity/graph/tool_node/base.py

async def stream(  # noqa: PLR0915
    self,
    name: str,
    args: dict,
    tool_call_id: str,
    config: dict[str, t.Any],
    state: AgentState,
    callback_manager: CallbackManager = Inject[CallbackManager],
) -> t.AsyncIterator[Message]:
    """Execute a tool with streaming support, yielding incremental results.

    Similar to invoke() but designed for tools that can provide streaming responses
    or when you want to process results as they become available. Currently,
    most tool providers return complete results, so this method typically yields
    a single Message with the full result.

    Args:
        name: The name of the tool to execute.
        args: Dictionary of arguments to pass to the tool function.
        tool_call_id: Unique identifier for this tool execution.
        config: Configuration dictionary containing execution context.
        state: Current agent state for context-aware tool execution.
        callback_manager: Manager for executing pre/post execution callbacks.

    Yields:
        Message objects containing tool execution results or status updates.
        For most tools, this will yield a single complete result Message.

    Example:
        ```python
        async for message in tool_node.stream(
            name="data_processor",
            args={"dataset": "large_data.csv"},
            tool_call_id="call_stream123",
            config={"user_id": "user1"},
            state=current_state,
        ):
            print(f"Received: {message.content}")
            # Process each streamed result
        ```

    Note:
        The streaming interface is designed for future expansion where tools
        may provide true streaming responses. Currently, it provides a
        consistent async iterator interface over tool results.
    """
    logger.info("Executing tool '%s' with %d arguments", name, len(args))
    logger.debug("Tool arguments: %s", args)
    event = EventModel.default(
        config,
        data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
        content_type=[ContentType.TOOL_CALL],
        event=Event.TOOL_EXECUTION,
    )
    event.node_name = "ToolNode"
    with contextlib.suppress(Exception):
        event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]

    if name in self.mcp_tools:
        event.metadata["function_type"] = "mcp"
        publish_event(event)
        message = await self._mcp_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self.composio_tools:
        event.metadata["function_type"] = "composio"
        publish_event(event)
        message = await self._composio_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self.langchain_tools:
        event.metadata["function_type"] = "langchain"
        publish_event(event)
        message = await self._langchain_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self._funcs:
        event.metadata["function_type"] = "internal"
        publish_event(event)

        result = await self._internal_execute(
            name,
            args,
            tool_call_id,
            config,
            state,
            callback_manager,
        )
        event.data["message"] = result.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=result.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield result
        return

    error_msg = f"Tool '{name}' not found."
    event.data["error"] = error_msg
    event.event_type = EventType.ERROR
    event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
    publish_event(event)

    yield Message.tool_message(
        content=[
            ErrorBlock(message=error_msg),
            ToolResultBlock(
                call_id=tool_call_id,
                output=error_msg,
                is_error=True,
                status="failed",
            ),
        ],
    )

Modules

base

Tool execution node for PyAgenity graph workflows.

This module provides the ToolNode class, which serves as a unified registry and executor for callable functions from various sources including local functions, MCP (Model Context Protocol) tools, Composio adapters, and LangChain tools. The ToolNode is designed with a modular architecture using mixins to handle different tool providers.

The ToolNode maintains compatibility with PyAgenity's dependency injection system and publishes execution events for monitoring and debugging purposes.

Typical usage example

def my_tool(query: str) -> str:
    return f"Result for: {query}"


tools = ToolNode([my_tool])
result = await tools.invoke("my_tool", {"query": "test"}, "call_id", config, state)

Classes:

Name	Description
ToolNode	A unified registry and executor for callable functions from various tool providers.

Attributes:

Name	Type	Description
logger

Attributes

logger module-attribute

logger = getLogger(__name__)

Classes

ToolNode

Bases: SchemaMixin, LocalExecMixin, MCPMixin, ComposioMixin, LangChainMixin, KwargsResolverMixin

A unified registry and executor for callable functions from various tool providers.

ToolNode serves as the central hub for managing and executing tools from multiple sources: - Local Python functions - MCP (Model Context Protocol) tools - Composio adapter tools - LangChain tools

The class uses a mixin-based architecture to separate concerns and maintain clean integration with different tool providers. It provides both synchronous and asynchronous execution methods with comprehensive event publishing and error handling.

Attributes:

Name	Type	Description
_funcs	dict[str, Callable]	Dictionary mapping function names to callable functions.
_client	Client | None	Optional MCP client for remote tool execution.
_composio	ComposioAdapter | None	Optional Composio adapter for external integrations.
_langchain	Any | None	Optional LangChain adapter for LangChain tools.
mcp_tools	list[str]	List of available MCP tool names.
composio_tools	list[str]	List of available Composio tool names.
langchain_tools	list[str]	List of available LangChain tool names.

Example

# Define local tools
def weather_tool(location: str) -> str:
    return f"Weather in {location}: Sunny, 25°C"


def calculator(a: int, b: int) -> int:
    return a + b


# Create ToolNode with local functions
tools = ToolNode([weather_tool, calculator])

# Execute a tool
result = await tools.invoke(
    name="weather_tool",
    args={"location": "New York"},
    tool_call_id="call_123",
    config={"user_id": "user1"},
    state=agent_state,
)

Methods:

Name	Description
__init__	Initialize ToolNode with functions and optional tool adapters.
all_tools	Get all available tools from all configured providers.
all_tools_sync	Synchronously get all available tools from all configured providers.
get_local_tool	Generate OpenAI-compatible tool definitions for all registered local functions.
invoke	Execute a specific tool by name with the provided arguments.
stream	Execute a tool with streaming support, yielding incremental results.

Source code in pyagenity/graph/tool_node/base.py

class ToolNode(
    SchemaMixin,
    LocalExecMixin,
    MCPMixin,
    ComposioMixin,
    LangChainMixin,
    KwargsResolverMixin,
):
    """A unified registry and executor for callable functions from various tool providers.

    ToolNode serves as the central hub for managing and executing tools from multiple sources:
    - Local Python functions
    - MCP (Model Context Protocol) tools
    - Composio adapter tools
    - LangChain tools

    The class uses a mixin-based architecture to separate concerns and maintain clean
    integration with different tool providers. It provides both synchronous and asynchronous
    execution methods with comprehensive event publishing and error handling.

    Attributes:
        _funcs: Dictionary mapping function names to callable functions.
        _client: Optional MCP client for remote tool execution.
        _composio: Optional Composio adapter for external integrations.
        _langchain: Optional LangChain adapter for LangChain tools.
        mcp_tools: List of available MCP tool names.
        composio_tools: List of available Composio tool names.
        langchain_tools: List of available LangChain tool names.

    Example:
        ```python
        # Define local tools
        def weather_tool(location: str) -> str:
            return f"Weather in {location}: Sunny, 25°C"


        def calculator(a: int, b: int) -> int:
            return a + b


        # Create ToolNode with local functions
        tools = ToolNode([weather_tool, calculator])

        # Execute a tool
        result = await tools.invoke(
            name="weather_tool",
            args={"location": "New York"},
            tool_call_id="call_123",
            config={"user_id": "user1"},
            state=agent_state,
        )
        ```
    """

    def __init__(
        self,
        functions: t.Iterable[t.Callable],
        client: deps.Client | None = None,  # type: ignore
        composio_adapter: ComposioAdapter | None = None,
        langchain_adapter: t.Any | None = None,
    ) -> None:
        """Initialize ToolNode with functions and optional tool adapters.

        Args:
            functions: Iterable of callable functions to register as tools. Each function
                will be registered with its `__name__` as the tool identifier.
            client: Optional MCP (Model Context Protocol) client for remote tool access.
                Requires 'fastmcp' and 'mcp' packages to be installed.
            composio_adapter: Optional Composio adapter for external integrations and
                third-party API access.
            langchain_adapter: Optional LangChain adapter for accessing LangChain tools
                and integrations.

        Raises:
            ImportError: If MCP client is provided but required packages are not installed.
            TypeError: If any item in functions is not callable.

        Note:
            When using MCP client functionality, ensure you have installed the required
            dependencies with: `pip install pyagenity[mcp]`
        """
        logger.info("Initializing ToolNode with %d functions", len(list(functions)))

        if client is not None:
            # Read flags dynamically so tests can patch pyagenity.graph.tool_node.HAS_*
            mod = sys.modules.get("pyagenity.graph.tool_node")
            has_fastmcp = getattr(mod, "HAS_FASTMCP", deps.HAS_FASTMCP) if mod else deps.HAS_FASTMCP
            has_mcp = getattr(mod, "HAS_MCP", deps.HAS_MCP) if mod else deps.HAS_MCP

            if not has_fastmcp or not has_mcp:
                raise ImportError(
                    "MCP client functionality requires 'fastmcp' and 'mcp' packages. "
                    "Install with: pip install pyagenity[mcp]"
                )
            logger.debug("ToolNode initialized with MCP client")

        self._funcs: dict[str, t.Callable] = {}
        self._client: deps.Client | None = client  # type: ignore
        self._composio: ComposioAdapter | None = composio_adapter
        self._langchain: t.Any | None = langchain_adapter

        for fn in functions:
            if not callable(fn):
                raise TypeError("ToolNode only accepts callables")
            self._funcs[fn.__name__] = fn

        self.mcp_tools: list[str] = []
        self.composio_tools: list[str] = []
        self.langchain_tools: list[str] = []

    async def _all_tools_async(self) -> list[dict]:
        tools: list[dict] = self.get_local_tool()
        tools.extend(await self._get_mcp_tool())
        tools.extend(await self._get_composio_tools())
        tools.extend(await self._get_langchain_tools())
        return tools

    async def all_tools(self) -> list[dict]:
        """Get all available tools from all configured providers.

        Retrieves and combines tool definitions from local functions, MCP client,
        Composio adapter, and LangChain adapter. Each tool definition includes
        the function schema with parameters and descriptions.

        Returns:
            List of tool definitions in OpenAI function calling format. Each dict
            contains 'type': 'function' and 'function' with name, description,
            and parameters schema.

        Example:
            ```python
            tools = await tool_node.all_tools()
            # Returns:
            # [
            #   {
            #     "type": "function",
            #     "function": {
            #       "name": "weather_tool",
            #       "description": "Get weather information for a location",
            #       "parameters": {
            #         "type": "object",
            #         "properties": {
            #           "location": {"type": "string"}
            #         },
            #         "required": ["location"]
            #       }
            #     }
            #   }
            # ]
            ```
        """
        return await self._all_tools_async()

    def all_tools_sync(self) -> list[dict]:
        """Synchronously get all available tools from all configured providers.

        This is a synchronous wrapper around the async all_tools() method.
        It uses asyncio.run() to handle async operations from MCP, Composio,
        and LangChain adapters.

        Returns:
            List of tool definitions in OpenAI function calling format.

        Note:
            Prefer using the async `all_tools()` method when possible, especially
            in async contexts, to avoid potential event loop issues.
        """
        tools: list[dict] = self.get_local_tool()
        if self._client:
            result = asyncio.run(self._get_mcp_tool())
            if result:
                tools.extend(result)
        comp = asyncio.run(self._get_composio_tools())
        if comp:
            tools.extend(comp)
        lc = asyncio.run(self._get_langchain_tools())
        if lc:
            tools.extend(lc)
        return tools

    async def invoke(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        state: AgentState,
        callback_manager: CallbackManager = Inject[CallbackManager],
    ) -> t.Any:
        """Execute a specific tool by name with the provided arguments.

        This method handles tool execution across all configured providers (local,
        MCP, Composio, LangChain) with comprehensive error handling, event publishing,
        and callback management.

        Args:
            name: The name of the tool to execute.
            args: Dictionary of arguments to pass to the tool function.
            tool_call_id: Unique identifier for this tool execution, used for
                tracking and result correlation.
            config: Configuration dictionary containing execution context and
                user-specific settings.
            state: Current agent state for context-aware tool execution.
            callback_manager: Manager for executing pre/post execution callbacks.
                Injected via dependency injection if not provided.

        Returns:
            Message object containing tool execution results, either successful
            output or error information with appropriate status indicators.

        Raises:
            The method handles all exceptions internally and returns error Messages
            rather than raising exceptions, ensuring robust execution flow.

        Example:
            ```python
            result = await tool_node.invoke(
                name="weather_tool",
                args={"location": "Paris", "units": "metric"},
                tool_call_id="call_abc123",
                config={"user_id": "user1", "session_id": "session1"},
                state=current_agent_state,
            )

            # result is a Message with tool execution results
            print(result.content)  # Tool output or error information
            ```

        Note:
            The method publishes execution events throughout the process for
            monitoring and debugging purposes. Tool execution is routed based
            on tool provider precedence: MCP → Composio → LangChain → Local.
        """
        logger.info("Executing tool '%s' with %d arguments", name, len(args))
        logger.debug("Tool arguments: %s", args)

        event = EventModel.default(
            config,
            data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.node_name = name
        # Attach structured tool call block
        with contextlib.suppress(Exception):
            event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]
        publish_event(event)

        if name in self.mcp_tools:
            event.metadata["is_mcp"] = True
            publish_event(event)
            res = await self._mcp_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            # Attach tool result block mirroring the tool output
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self.composio_tools:
            event.metadata["is_composio"] = True
            publish_event(event)
            res = await self._composio_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self.langchain_tools:
            event.metadata["is_langchain"] = True
            publish_event(event)
            res = await self._langchain_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        if name in self._funcs:
            event.metadata["is_mcp"] = False
            publish_event(event)
            res = await self._internal_execute(
                name,
                args,
                tool_call_id,
                config,
                state,
                callback_manager,
            )
            event.data["message"] = res.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res

        error_msg = f"Tool '{name}' not found."
        event.data["error"] = error_msg
        event.event_type = EventType.ERROR
        event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
        publish_event(event)
        return Message.tool_message(
            content=[
                ErrorBlock(message=error_msg),
                ToolResultBlock(
                    call_id=tool_call_id,
                    output=error_msg,
                    is_error=True,
                    status="failed",
                ),
            ],
        )

    async def stream(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        state: AgentState,
        callback_manager: CallbackManager = Inject[CallbackManager],
    ) -> t.AsyncIterator[Message]:
        """Execute a tool with streaming support, yielding incremental results.

        Similar to invoke() but designed for tools that can provide streaming responses
        or when you want to process results as they become available. Currently,
        most tool providers return complete results, so this method typically yields
        a single Message with the full result.

        Args:
            name: The name of the tool to execute.
            args: Dictionary of arguments to pass to the tool function.
            tool_call_id: Unique identifier for this tool execution.
            config: Configuration dictionary containing execution context.
            state: Current agent state for context-aware tool execution.
            callback_manager: Manager for executing pre/post execution callbacks.

        Yields:
            Message objects containing tool execution results or status updates.
            For most tools, this will yield a single complete result Message.

        Example:
            ```python
            async for message in tool_node.stream(
                name="data_processor",
                args={"dataset": "large_data.csv"},
                tool_call_id="call_stream123",
                config={"user_id": "user1"},
                state=current_state,
            ):
                print(f"Received: {message.content}")
                # Process each streamed result
            ```

        Note:
            The streaming interface is designed for future expansion where tools
            may provide true streaming responses. Currently, it provides a
            consistent async iterator interface over tool results.
        """
        logger.info("Executing tool '%s' with %d arguments", name, len(args))
        logger.debug("Tool arguments: %s", args)
        event = EventModel.default(
            config,
            data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.node_name = "ToolNode"
        with contextlib.suppress(Exception):
            event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]

        if name in self.mcp_tools:
            event.metadata["function_type"] = "mcp"
            publish_event(event)
            message = await self._mcp_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self.composio_tools:
            event.metadata["function_type"] = "composio"
            publish_event(event)
            message = await self._composio_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self.langchain_tools:
            event.metadata["function_type"] = "langchain"
            publish_event(event)
            message = await self._langchain_execute(
                name,
                args,
                tool_call_id,
                config,
                callback_manager,
            )
            event.data["message"] = message.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield message
            return

        if name in self._funcs:
            event.metadata["function_type"] = "internal"
            publish_event(event)

            result = await self._internal_execute(
                name,
                args,
                tool_call_id,
                config,
                state,
                callback_manager,
            )
            event.data["message"] = result.model_dump()
            with contextlib.suppress(Exception):
                event.content_blocks = [
                    ToolResultBlock(call_id=tool_call_id, output=result.model_dump())
                ]
            event.event_type = EventType.END
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            yield result
            return

        error_msg = f"Tool '{name}' not found."
        event.data["error"] = error_msg
        event.event_type = EventType.ERROR
        event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
        publish_event(event)

        yield Message.tool_message(
            content=[
                ErrorBlock(message=error_msg),
                ToolResultBlock(
                    call_id=tool_call_id,
                    output=error_msg,
                    is_error=True,
                    status="failed",
                ),
            ],
        )

Attributes

composio_tools instance-attribute

composio_tools = []

langchain_tools instance-attribute

langchain_tools = []

mcp_tools instance-attribute

mcp_tools = []

Functions

__init__

__init__(functions, client=None, composio_adapter=None, langchain_adapter=None)

Initialize ToolNode with functions and optional tool adapters.

Parameters:

Name	Type	Description	Default
functions	Iterable[Callable]	Iterable of callable functions to register as tools. Each function will be registered with its __name__ as the tool identifier.	required
client	Client | None	Optional MCP (Model Context Protocol) client for remote tool access. Requires 'fastmcp' and 'mcp' packages to be installed.	None
composio_adapter	ComposioAdapter | None	Optional Composio adapter for external integrations and third-party API access.	None
langchain_adapter	Any | None	Optional LangChain adapter for accessing LangChain tools and integrations.	None

Raises:

Type	Description
ImportError	If MCP client is provided but required packages are not installed.
TypeError	If any item in functions is not callable.

Note

When using MCP client functionality, ensure you have installed the required dependencies with: pip install pyagenity[mcp]

Source code in pyagenity/graph/tool_node/base.py

def __init__(
    self,
    functions: t.Iterable[t.Callable],
    client: deps.Client | None = None,  # type: ignore
    composio_adapter: ComposioAdapter | None = None,
    langchain_adapter: t.Any | None = None,
) -> None:
    """Initialize ToolNode with functions and optional tool adapters.

    Args:
        functions: Iterable of callable functions to register as tools. Each function
            will be registered with its `__name__` as the tool identifier.
        client: Optional MCP (Model Context Protocol) client for remote tool access.
            Requires 'fastmcp' and 'mcp' packages to be installed.
        composio_adapter: Optional Composio adapter for external integrations and
            third-party API access.
        langchain_adapter: Optional LangChain adapter for accessing LangChain tools
            and integrations.

    Raises:
        ImportError: If MCP client is provided but required packages are not installed.
        TypeError: If any item in functions is not callable.

    Note:
        When using MCP client functionality, ensure you have installed the required
        dependencies with: `pip install pyagenity[mcp]`
    """
    logger.info("Initializing ToolNode with %d functions", len(list(functions)))

    if client is not None:
        # Read flags dynamically so tests can patch pyagenity.graph.tool_node.HAS_*
        mod = sys.modules.get("pyagenity.graph.tool_node")
        has_fastmcp = getattr(mod, "HAS_FASTMCP", deps.HAS_FASTMCP) if mod else deps.HAS_FASTMCP
        has_mcp = getattr(mod, "HAS_MCP", deps.HAS_MCP) if mod else deps.HAS_MCP

        if not has_fastmcp or not has_mcp:
            raise ImportError(
                "MCP client functionality requires 'fastmcp' and 'mcp' packages. "
                "Install with: pip install pyagenity[mcp]"
            )
        logger.debug("ToolNode initialized with MCP client")

    self._funcs: dict[str, t.Callable] = {}
    self._client: deps.Client | None = client  # type: ignore
    self._composio: ComposioAdapter | None = composio_adapter
    self._langchain: t.Any | None = langchain_adapter

    for fn in functions:
        if not callable(fn):
            raise TypeError("ToolNode only accepts callables")
        self._funcs[fn.__name__] = fn

    self.mcp_tools: list[str] = []
    self.composio_tools: list[str] = []
    self.langchain_tools: list[str] = []

all_tools async

all_tools()

Get all available tools from all configured providers.

Retrieves and combines tool definitions from local functions, MCP client, Composio adapter, and LangChain adapter. Each tool definition includes the function schema with parameters and descriptions.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format. Each dict
list[dict]	contains 'type': 'function' and 'function' with name, description,
list[dict]	and parameters schema.

Example

tools = await tool_node.all_tools()
# Returns:
# [
#   {
#     "type": "function",
#     "function": {
#       "name": "weather_tool",
#       "description": "Get weather information for a location",
#       "parameters": {
#         "type": "object",
#         "properties": {
#           "location": {"type": "string"}
#         },
#         "required": ["location"]
#       }
#     }
#   }
# ]

Source code in pyagenity/graph/tool_node/base.py

async def all_tools(self) -> list[dict]:
    """Get all available tools from all configured providers.

    Retrieves and combines tool definitions from local functions, MCP client,
    Composio adapter, and LangChain adapter. Each tool definition includes
    the function schema with parameters and descriptions.

    Returns:
        List of tool definitions in OpenAI function calling format. Each dict
        contains 'type': 'function' and 'function' with name, description,
        and parameters schema.

    Example:
        ```python
        tools = await tool_node.all_tools()
        # Returns:
        # [
        #   {
        #     "type": "function",
        #     "function": {
        #       "name": "weather_tool",
        #       "description": "Get weather information for a location",
        #       "parameters": {
        #         "type": "object",
        #         "properties": {
        #           "location": {"type": "string"}
        #         },
        #         "required": ["location"]
        #       }
        #     }
        #   }
        # ]
        ```
    """
    return await self._all_tools_async()

all_tools_sync

all_tools_sync()

Synchronously get all available tools from all configured providers.

This is a synchronous wrapper around the async all_tools() method. It uses asyncio.run() to handle async operations from MCP, Composio, and LangChain adapters.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format.

Note

Prefer using the async all_tools() method when possible, especially in async contexts, to avoid potential event loop issues.

Source code in pyagenity/graph/tool_node/base.py

def all_tools_sync(self) -> list[dict]:
    """Synchronously get all available tools from all configured providers.

    This is a synchronous wrapper around the async all_tools() method.
    It uses asyncio.run() to handle async operations from MCP, Composio,
    and LangChain adapters.

    Returns:
        List of tool definitions in OpenAI function calling format.

    Note:
        Prefer using the async `all_tools()` method when possible, especially
        in async contexts, to avoid potential event loop issues.
    """
    tools: list[dict] = self.get_local_tool()
    if self._client:
        result = asyncio.run(self._get_mcp_tool())
        if result:
            tools.extend(result)
    comp = asyncio.run(self._get_composio_tools())
    if comp:
        tools.extend(comp)
    lc = asyncio.run(self._get_langchain_tools())
    if lc:
        tools.extend(lc)
    return tools

get_local_tool

get_local_tool()

Generate OpenAI-compatible tool definitions for all registered local functions.

Inspects all registered functions in _funcs and automatically generates tool schemas by analyzing function signatures, type annotations, and docstrings. Excludes injectable parameters that are provided by the framework.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format. Each
list[dict]	definition includes the function name, description (from docstring),
list[dict]	and complete parameter schema with types and required fields.

Example

For a function:

def calculate(a: int, b: int, operation: str = "add") -> int:
    '''Perform arithmetic calculation.'''
    return a + b if operation == "add" else a - b

Returns:

[
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Perform arithmetic calculation.",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "integer"},
                    "b": {"type": "integer"},
                    "operation": {"type": "string", "default": "add"},
                },
                "required": ["a", "b"],
            },
        },
    }
]

Note

Parameters listed in INJECTABLE_PARAMS (like 'state', 'config', 'tool_call_id') are automatically excluded from the generated schema as they are provided by the framework during execution.

Source code in pyagenity/graph/tool_node/schema.py

def get_local_tool(self) -> list[dict]:
    """Generate OpenAI-compatible tool definitions for all registered local functions.

    Inspects all registered functions in _funcs and automatically generates
    tool schemas by analyzing function signatures, type annotations, and docstrings.
    Excludes injectable parameters that are provided by the framework.

    Returns:
        List of tool definitions in OpenAI function calling format. Each
        definition includes the function name, description (from docstring),
        and complete parameter schema with types and required fields.

    Example:
        For a function:
        ```python
        def calculate(a: int, b: int, operation: str = "add") -> int:
            '''Perform arithmetic calculation.'''
            return a + b if operation == "add" else a - b
        ```

        Returns:
        ```python
        [
            {
                "type": "function",
                "function": {
                    "name": "calculate",
                    "description": "Perform arithmetic calculation.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "a": {"type": "integer"},
                            "b": {"type": "integer"},
                            "operation": {"type": "string", "default": "add"},
                        },
                        "required": ["a", "b"],
                    },
                },
            }
        ]
        ```

    Note:
        Parameters listed in INJECTABLE_PARAMS (like 'state', 'config',
        'tool_call_id') are automatically excluded from the generated schema
        as they are provided by the framework during execution.
    """
    tools: list[dict] = []
    for name, fn in self._funcs.items():
        sig = inspect.signature(fn)
        params_schema: dict = {"type": "object", "properties": {}, "required": []}

        for p_name, p in sig.parameters.items():
            if p.kind in (
                inspect.Parameter.VAR_POSITIONAL,
                inspect.Parameter.VAR_KEYWORD,
            ):
                continue

            if p_name in INJECTABLE_PARAMS:
                continue

            annotation = p.annotation if p.annotation is not inspect._empty else str
            prop = SchemaMixin._annotation_to_schema(annotation, p.default)
            params_schema["properties"][p_name] = prop

            if p.default is inspect._empty:
                params_schema["required"].append(p_name)

        if not params_schema["required"]:
            params_schema.pop("required")

        description = inspect.getdoc(fn) or "No description provided."

        # provider = getattr(fn, "_py_tool_provider", None)
        # tags = getattr(fn, "_py_tool_tags", None)
        # capabilities = getattr(fn, "_py_tool_capabilities", None)

        entry = {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": params_schema,
            },
        }
        # meta: dict[str, t.Any] = {}
        # if provider:
        #     meta["provider"] = provider
        # if tags:
        #     meta["tags"] = tags
        # if capabilities:
        #     meta["capabilities"] = capabilities
        # if meta:
        #     entry["x-pyagenity"] = meta

        tools.append(entry)

    return tools

invoke async

invoke(name, args, tool_call_id, config, state, callback_manager=Inject[CallbackManager])

Execute a specific tool by name with the provided arguments.

This method handles tool execution across all configured providers (local, MCP, Composio, LangChain) with comprehensive error handling, event publishing, and callback management.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to execute.	required
args	dict	Dictionary of arguments to pass to the tool function.	required
tool_call_id	str	Unique identifier for this tool execution, used for tracking and result correlation.	required
config	dict[str, Any]	Configuration dictionary containing execution context and user-specific settings.	required
state	AgentState	Current agent state for context-aware tool execution.	required
callback_manager	CallbackManager	Manager for executing pre/post execution callbacks. Injected via dependency injection if not provided.	Inject[CallbackManager]

Returns:

Type	Description
Any	Message object containing tool execution results, either successful
Any	output or error information with appropriate status indicators.

Example

result = await tool_node.invoke(
    name="weather_tool",
    args={"location": "Paris", "units": "metric"},
    tool_call_id="call_abc123",
    config={"user_id": "user1", "session_id": "session1"},
    state=current_agent_state,
)

# result is a Message with tool execution results
print(result.content)  # Tool output or error information

Note

The method publishes execution events throughout the process for monitoring and debugging purposes. Tool execution is routed based on tool provider precedence: MCP → Composio → LangChain → Local.

Source code in pyagenity/graph/tool_node/base.py

async def invoke(  # noqa: PLR0915
    self,
    name: str,
    args: dict,
    tool_call_id: str,
    config: dict[str, t.Any],
    state: AgentState,
    callback_manager: CallbackManager = Inject[CallbackManager],
) -> t.Any:
    """Execute a specific tool by name with the provided arguments.

    This method handles tool execution across all configured providers (local,
    MCP, Composio, LangChain) with comprehensive error handling, event publishing,
    and callback management.

    Args:
        name: The name of the tool to execute.
        args: Dictionary of arguments to pass to the tool function.
        tool_call_id: Unique identifier for this tool execution, used for
            tracking and result correlation.
        config: Configuration dictionary containing execution context and
            user-specific settings.
        state: Current agent state for context-aware tool execution.
        callback_manager: Manager for executing pre/post execution callbacks.
            Injected via dependency injection if not provided.

    Returns:
        Message object containing tool execution results, either successful
        output or error information with appropriate status indicators.

    Raises:
        The method handles all exceptions internally and returns error Messages
        rather than raising exceptions, ensuring robust execution flow.

    Example:
        ```python
        result = await tool_node.invoke(
            name="weather_tool",
            args={"location": "Paris", "units": "metric"},
            tool_call_id="call_abc123",
            config={"user_id": "user1", "session_id": "session1"},
            state=current_agent_state,
        )

        # result is a Message with tool execution results
        print(result.content)  # Tool output or error information
        ```

    Note:
        The method publishes execution events throughout the process for
        monitoring and debugging purposes. Tool execution is routed based
        on tool provider precedence: MCP → Composio → LangChain → Local.
    """
    logger.info("Executing tool '%s' with %d arguments", name, len(args))
    logger.debug("Tool arguments: %s", args)

    event = EventModel.default(
        config,
        data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
        content_type=[ContentType.TOOL_CALL],
        event=Event.TOOL_EXECUTION,
    )
    event.node_name = name
    # Attach structured tool call block
    with contextlib.suppress(Exception):
        event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]
    publish_event(event)

    if name in self.mcp_tools:
        event.metadata["is_mcp"] = True
        publish_event(event)
        res = await self._mcp_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        # Attach tool result block mirroring the tool output
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self.composio_tools:
        event.metadata["is_composio"] = True
        publish_event(event)
        res = await self._composio_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self.langchain_tools:
        event.metadata["is_langchain"] = True
        publish_event(event)
        res = await self._langchain_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    if name in self._funcs:
        event.metadata["is_mcp"] = False
        publish_event(event)
        res = await self._internal_execute(
            name,
            args,
            tool_call_id,
            config,
            state,
            callback_manager,
        )
        event.data["message"] = res.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=res.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        return res

    error_msg = f"Tool '{name}' not found."
    event.data["error"] = error_msg
    event.event_type = EventType.ERROR
    event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
    publish_event(event)
    return Message.tool_message(
        content=[
            ErrorBlock(message=error_msg),
            ToolResultBlock(
                call_id=tool_call_id,
                output=error_msg,
                is_error=True,
                status="failed",
            ),
        ],
    )

stream async

stream(name, args, tool_call_id, config, state, callback_manager=Inject[CallbackManager])

Execute a tool with streaming support, yielding incremental results.

Similar to invoke() but designed for tools that can provide streaming responses or when you want to process results as they become available. Currently, most tool providers return complete results, so this method typically yields a single Message with the full result.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to execute.	required
args	dict	Dictionary of arguments to pass to the tool function.	required
tool_call_id	str	Unique identifier for this tool execution.	required
config	dict[str, Any]	Configuration dictionary containing execution context.	required
state	AgentState	Current agent state for context-aware tool execution.	required
callback_manager	CallbackManager	Manager for executing pre/post execution callbacks.	Inject[CallbackManager]

Yields:

Type	Description
AsyncIterator[Message]	Message objects containing tool execution results or status updates.
AsyncIterator[Message]	For most tools, this will yield a single complete result Message.

Example

async for message in tool_node.stream(
    name="data_processor",
    args={"dataset": "large_data.csv"},
    tool_call_id="call_stream123",
    config={"user_id": "user1"},
    state=current_state,
):
    print(f"Received: {message.content}")
    # Process each streamed result

Note

The streaming interface is designed for future expansion where tools may provide true streaming responses. Currently, it provides a consistent async iterator interface over tool results.

Source code in pyagenity/graph/tool_node/base.py

async def stream(  # noqa: PLR0915
    self,
    name: str,
    args: dict,
    tool_call_id: str,
    config: dict[str, t.Any],
    state: AgentState,
    callback_manager: CallbackManager = Inject[CallbackManager],
) -> t.AsyncIterator[Message]:
    """Execute a tool with streaming support, yielding incremental results.

    Similar to invoke() but designed for tools that can provide streaming responses
    or when you want to process results as they become available. Currently,
    most tool providers return complete results, so this method typically yields
    a single Message with the full result.

    Args:
        name: The name of the tool to execute.
        args: Dictionary of arguments to pass to the tool function.
        tool_call_id: Unique identifier for this tool execution.
        config: Configuration dictionary containing execution context.
        state: Current agent state for context-aware tool execution.
        callback_manager: Manager for executing pre/post execution callbacks.

    Yields:
        Message objects containing tool execution results or status updates.
        For most tools, this will yield a single complete result Message.

    Example:
        ```python
        async for message in tool_node.stream(
            name="data_processor",
            args={"dataset": "large_data.csv"},
            tool_call_id="call_stream123",
            config={"user_id": "user1"},
            state=current_state,
        ):
            print(f"Received: {message.content}")
            # Process each streamed result
        ```

    Note:
        The streaming interface is designed for future expansion where tools
        may provide true streaming responses. Currently, it provides a
        consistent async iterator interface over tool results.
    """
    logger.info("Executing tool '%s' with %d arguments", name, len(args))
    logger.debug("Tool arguments: %s", args)
    event = EventModel.default(
        config,
        data={"args": args, "tool_call_id": tool_call_id, "function_name": name},
        content_type=[ContentType.TOOL_CALL],
        event=Event.TOOL_EXECUTION,
    )
    event.node_name = "ToolNode"
    with contextlib.suppress(Exception):
        event.content_blocks = [ToolCallBlock(id=tool_call_id, name=name, args=args)]

    if name in self.mcp_tools:
        event.metadata["function_type"] = "mcp"
        publish_event(event)
        message = await self._mcp_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self.composio_tools:
        event.metadata["function_type"] = "composio"
        publish_event(event)
        message = await self._composio_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self.langchain_tools:
        event.metadata["function_type"] = "langchain"
        publish_event(event)
        message = await self._langchain_execute(
            name,
            args,
            tool_call_id,
            config,
            callback_manager,
        )
        event.data["message"] = message.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=message.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield message
        return

    if name in self._funcs:
        event.metadata["function_type"] = "internal"
        publish_event(event)

        result = await self._internal_execute(
            name,
            args,
            tool_call_id,
            config,
            state,
            callback_manager,
        )
        event.data["message"] = result.model_dump()
        with contextlib.suppress(Exception):
            event.content_blocks = [
                ToolResultBlock(call_id=tool_call_id, output=result.model_dump())
            ]
        event.event_type = EventType.END
        event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
        publish_event(event)
        yield result
        return

    error_msg = f"Tool '{name}' not found."
    event.data["error"] = error_msg
    event.event_type = EventType.ERROR
    event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
    publish_event(event)

    yield Message.tool_message(
        content=[
            ErrorBlock(message=error_msg),
            ToolResultBlock(
                call_id=tool_call_id,
                output=error_msg,
                is_error=True,
                status="failed",
            ),
        ],
    )

Functions

Modules

constants

Constants for ToolNode package.

This module defines constants used throughout the ToolNode implementation, particularly parameter names that are automatically injected by the PyAgenity framework during tool execution. These parameters are excluded from tool schema generation since they are provided by the execution context.

The constants are separated into their own module to avoid circular imports and maintain a clean public API.

Parameter names that are automatically injected during tool execution.

These parameters are provided by the PyAgenity framework and should be excluded from tool schema generation. They represent execution context and framework services that are available to tool functions but not provided by the user.

Parameters:

Name	Type	Description	Default
tool_call_id		Unique identifier for the current tool execution.	required
state		Current AgentState instance for context-aware execution.	required
config		Configuration dictionary with execution settings.	required
generated_id		Framework-generated identifier for various purposes.	required
context_manager		BaseContextManager instance for cross-node operations.	required
publisher		BasePublisher instance for event publishing.	required
checkpointer		BaseCheckpointer instance for state persistence.	required
store		BaseStore instance for data storage operations.	required

Note

Tool functions can declare these parameters in their signatures to receive the corresponding services, but they should not be included in the tool schema since they're not user-provided arguments.

Attributes:

Name	Type	Description
INJECTABLE_PARAMS

Attributes

INJECTABLE_PARAMS module-attribute

INJECTABLE_PARAMS = {'tool_call_id', 'state', 'config', 'generated_id', 'context_manager', 'publisher', 'checkpointer', 'store'}

deps

Dependency flags and optional imports for ToolNode.

This module manages optional third-party dependencies for the ToolNode implementation, providing clean import handling and feature flags. It isolates optional imports to prevent ImportError cascades when optional dependencies are not installed.

The module handles two main optional dependency groups: 1. MCP (Model Context Protocol) support via 'fastmcp' and 'mcp' packages 2. Future extensibility for other optional tool providers

By centralizing optional imports here, other modules can safely import the flags and types without triggering ImportError exceptions, allowing graceful degradation when optional features are not available.

Typical usage

from .deps import HAS_FASTMCP, HAS_MCP, Client

if HAS_FASTMCP and HAS_MCP:
    # Use MCP functionality
    client = Client(...)
else:
    # Graceful fallback or error message
    client = None

FastMCP integration support.

Boolean flag indicating whether FastMCP is available.

True if 'fastmcp' package is installed and imports successfully.

FastMCP Client class for connecting to MCP servers.

None if FastMCP is not available.

Result type for MCP tool executions.

None if FastMCP is not available.

Attributes:

Name	Type	Description
HAS_FASTMCP
HAS_MCP

Attributes

HAS_FASTMCP module-attribute

HAS_FASTMCP = True

HAS_MCP module-attribute

HAS_MCP = True

__all__ module-attribute

__all__ = ['HAS_FASTMCP', 'HAS_MCP', 'CallToolResult', 'Client', 'ContentBlock', 'Tool']

executors

Executors for different tool providers and local functions.

Classes:

Name	Description
ComposioMixin
KwargsResolverMixin
LangChainMixin
LocalExecMixin
MCPMixin

Attributes:

Name	Type	Description
logger

Attributes

logger module-attribute

logger = getLogger(__name__)

Classes

ComposioMixin

Attributes:

Name	Type	Description
composio_tools	list[str]

Source code in pyagenity/graph/tool_node/executors.py

class ComposioMixin:
    _composio: ComposioAdapter | None
    composio_tools: list[str]

    async def _get_composio_tools(self) -> list[dict]:
        tools: list[dict] = []
        if not self._composio:
            return tools
        try:
            raw = self._composio.list_raw_tools_for_llm()
            for tdef in raw:
                fn = tdef.get("function", {})
                name = fn.get("name")
                if name:
                    self.composio_tools.append(name)
                tools.append(tdef)
        except Exception as e:  # pragma: no cover - network/optional
            logger.exception("Failed to fetch Composio tools: %s", e)
        return tools

    async def _composio_execute(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        callback_mgr: CallbackManager,
    ) -> Message:
        context = CallbackContext(
            invocation_type=InvocationType.TOOL,
            node_name="ToolNode",
            function_name=name,
            metadata={
                "tool_call_id": tool_call_id,
                "args": args,
                "config": config,
                "composio": True,
            },
        )
        meta = {"function_name": name, "function_argument": args, "tool_call_id": tool_call_id}

        event = EventModel.default(
            base_config=config,
            data={
                "tool_call_id": tool_call_id,
                "args": args,
                "function_name": name,
                "is_composio": True,
            },
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.event_type = EventType.PROGRESS
        event.node_name = "ToolNode"
        event.sequence_id = 1
        publish_event(event)

        input_data = {**args}

        def safe_serialize(obj: t.Any) -> dict[str, t.Any]:
            try:
                json.dumps(obj)
                return obj if isinstance(obj, dict) else {"content": obj}
            except (TypeError, OverflowError):
                if hasattr(obj, "model_dump"):
                    dumped = obj.model_dump()  # type: ignore
                    if isinstance(dumped, dict) and dumped.get("type") == "resource":
                        resource = dumped.get("resource", {})
                        if isinstance(resource, dict) and "uri" in resource:
                            resource["uri"] = str(resource["uri"])
                            dumped["resource"] = resource
                    return dumped
                return {"content": str(obj), "type": "fallback"}

        try:
            input_data = await callback_mgr.execute_before_invoke(context, input_data)
            event.event_type = EventType.UPDATE
            event.sequence_id = 2
            event.metadata["status"] = "before_invoke_complete Invoke Composio"
            publish_event(event)

            comp_conf = (config.get("composio") if isinstance(config, dict) else None) or {}
            user_id = comp_conf.get("user_id") or config.get("user_id")
            connected_account_id = comp_conf.get("connected_account_id") or config.get(
                "connected_account_id"
            )

            if not self._composio:
                error_result = Message.tool_message(
                    content=[
                        ErrorBlock(message="Composio adapter not configured"),
                        ToolResultBlock(
                            call_id=tool_call_id,
                            output="Composio adapter not configured",
                            status="failed",
                            is_error=True,
                        ),
                    ],
                    meta=meta,
                )
                event.event_type = EventType.ERROR
                event.metadata["error"] = "Composio adapter not configured"
                publish_event(event)
                return error_result

            res = self._composio.execute(
                slug=name,
                arguments=input_data,
                user_id=user_id,
                connected_account_id=connected_account_id,
            )

            successful = bool(res.get("successful"))
            payload = res.get("data")
            error = res.get("error")

            result_blocks = []
            if error and not successful:
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output={"success": False, "error": error},
                        status="failed",
                        is_error=True,
                    )
                )
                result_blocks.append(ErrorBlock(message=error))
            else:
                if isinstance(payload, list):
                    output = [safe_serialize(item) for item in payload]
                else:
                    output = [safe_serialize(payload)]
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=output,
                        status="completed" if successful else "failed",
                        is_error=not successful,
                    )
                )

            result = Message.tool_message(
                content=result_blocks,
                meta=meta,
            )

            res_msg = await callback_mgr.execute_after_invoke(context, input_data, result)
            event.event_type = EventType.END
            event.data["message"] = result.model_dump()
            event.metadata["status"] = "Composio tool execution complete"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res_msg

        except Exception as e:  # pragma: no cover - error path
            recovery_result = await callback_mgr.execute_on_error(context, input_data, e)
            if isinstance(recovery_result, Message):
                event.event_type = EventType.END
                event.data["message"] = recovery_result.model_dump()
                event.metadata["status"] = "Composio tool execution complete, with recovery"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return recovery_result

            event.event_type = EventType.END
            event.data["error"] = str(e)
            event.metadata["status"] = "Composio tool execution complete, with error"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
            publish_event(event)
            return Message.tool_message(
                content=[
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=f"Composio execution error: {e}",
                        status="failed",
                        is_error=True,
                    ),
                    ErrorBlock(message=f"Composio execution error: {e}"),
                ],
                meta=meta,
            )

Attributes

composio_tools instance-attribute

composio_tools

KwargsResolverMixin

Source code in pyagenity/graph/tool_node/executors.py

class KwargsResolverMixin:
    def _should_skip_parameter(self, param: inspect.Parameter) -> bool:
        return param.kind in (
            inspect.Parameter.VAR_POSITIONAL,
            inspect.Parameter.VAR_KEYWORD,
        )

    def _handle_injectable_parameter(
        self,
        p_name: str,
        param: inspect.Parameter,
        injectable_params: dict,
        dependency_container,
    ) -> t.Any | None:
        if p_name in injectable_params:
            injectable_value = injectable_params[p_name]
            if injectable_value is not None:
                return injectable_value

        if dependency_container and dependency_container.has(p_name):
            return dependency_container.get(p_name)

        if param.default is inspect._empty:
            raise TypeError(f"Required injectable parameter '{p_name}' not found")

        return None

    def _get_parameter_value(
        self,
        p_name: str,
        param: inspect.Parameter,
        args: dict,
        injectable_params: dict,
        dependency_container,
    ) -> t.Any | None:
        if p_name in injectable_params:
            return self._handle_injectable_parameter(
                p_name, param, injectable_params, dependency_container
            )

        value_sources = [
            lambda: args.get(p_name),
            lambda: (
                dependency_container.get(p_name)
                if dependency_container and dependency_container.has(p_name)
                else None
            ),
        ]

        for source in value_sources:
            value = source()
            if value is not None:
                return value

        if param.default is not inspect._empty:
            return None

        raise TypeError(f"Missing required parameter '{p_name}' for function")

    def _prepare_kwargs(
        self,
        sig: inspect.Signature,
        args: dict,
        injectable_params: dict,
        dependency_container,
    ) -> dict:
        kwargs: dict = {}
        for p_name, p in sig.parameters.items():
            if self._should_skip_parameter(p):
                continue
            value = self._get_parameter_value(
                p_name, p, args, injectable_params, dependency_container
            )
            if value is not None:
                kwargs[p_name] = value
        return kwargs

LangChainMixin

Attributes:

Name	Type	Description
langchain_tools	list[str]

Source code in pyagenity/graph/tool_node/executors.py

class LangChainMixin:
    _langchain: t.Any | None
    langchain_tools: list[str]

    async def _get_langchain_tools(self) -> list[dict]:
        tools: list[dict] = []
        if not self._langchain:
            return tools
        try:
            raw = self._langchain.list_tools_for_llm()
            for tdef in raw:
                fn = tdef.get("function", {})
                name = fn.get("name")
                if name:
                    self.langchain_tools.append(name)
                tools.append(tdef)
        except Exception as e:  # pragma: no cover - optional
            logger.warning("Failed to fetch LangChain tools: %s", e)
        return tools

    async def _langchain_execute(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        callback_mgr: CallbackManager,
    ) -> Message:
        context = CallbackContext(
            invocation_type=InvocationType.TOOL,
            node_name="ToolNode",
            function_name=name,
            metadata={
                "tool_call_id": tool_call_id,
                "args": args,
                "config": config,
                "langchain": True,
            },
        )
        meta = {"function_name": name, "function_argument": args, "tool_call_id": tool_call_id}

        event = EventModel.default(
            base_config=config,
            data={
                "tool_call_id": tool_call_id,
                "args": args,
                "function_name": name,
                "is_langchain": True,
            },
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.event_type = EventType.PROGRESS
        event.node_name = "ToolNode"
        event.sequence_id = 1
        publish_event(event)

        input_data = {**args}

        def safe_serialize(obj: t.Any) -> dict[str, t.Any]:
            try:
                json.dumps(obj)
                return obj if isinstance(obj, dict) else {"content": obj}
            except (TypeError, OverflowError):
                if hasattr(obj, "model_dump"):
                    dumped = obj.model_dump()  # type: ignore
                    if isinstance(dumped, dict) and dumped.get("type") == "resource":
                        resource = dumped.get("resource", {})
                        if isinstance(resource, dict) and "uri" in resource:
                            resource["uri"] = str(resource["uri"])
                            dumped["resource"] = resource
                    return dumped
                return {"content": str(obj), "type": "fallback"}

        try:
            input_data = await callback_mgr.execute_before_invoke(context, input_data)
            event.event_type = EventType.UPDATE
            event.sequence_id = 2
            event.metadata["status"] = "before_invoke_complete Invoke LangChain"
            publish_event(event)

            if not self._langchain:
                error_result = Message.tool_message(
                    content=[
                        ErrorBlock(message="LangChain adapter not configured"),
                        ToolResultBlock(
                            call_id=tool_call_id,
                            output="LangChain adapter not configured",
                            status="failed",
                            is_error=True,
                        ),
                    ],
                    meta=meta,
                )
                event.event_type = EventType.ERROR
                event.metadata["error"] = "LangChain adapter not configured"
                publish_event(event)
                return error_result

            res = self._langchain.execute(name=name, arguments=input_data)
            successful = bool(res.get("successful"))
            payload = res.get("data")
            error = res.get("error")

            result_blocks = []
            if error and not successful:
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output={"success": False, "error": error},
                        status="failed",
                        is_error=True,
                    )
                )
                result_blocks.append(ErrorBlock(message=error))
            else:
                if isinstance(payload, list):
                    output = [safe_serialize(item) for item in payload]
                else:
                    output = [safe_serialize(payload)]
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=output,
                        status="completed" if successful else "failed",
                        is_error=not successful,
                    )
                )

            result = Message.tool_message(
                content=result_blocks,
                meta=meta,
            )

            res_msg = await callback_mgr.execute_after_invoke(context, input_data, result)
            event.event_type = EventType.END
            event.data["message"] = result.model_dump()
            event.metadata["status"] = "LangChain tool execution complete"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)
            return res_msg

        except Exception as e:  # pragma: no cover - error path
            recovery_result = await callback_mgr.execute_on_error(context, input_data, e)
            if isinstance(recovery_result, Message):
                event.event_type = EventType.END
                event.data["message"] = recovery_result.model_dump()
                event.metadata["status"] = "LangChain tool execution complete, with recovery"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return recovery_result

            event.event_type = EventType.END
            event.data["error"] = str(e)
            event.metadata["status"] = "LangChain tool execution complete, with error"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
            publish_event(event)
            return Message.tool_message(
                content=[
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=f"LangChain execution error: {e}",
                        status="failed",
                        is_error=True,
                    ),
                    ErrorBlock(message=f"LangChain execution error: {e}"),
                ],
                meta=meta,
            )

Attributes

langchain_tools instance-attribute

langchain_tools

LocalExecMixin

Source code in pyagenity/graph/tool_node/executors.py

class LocalExecMixin:
    _funcs: dict[str, t.Callable]

    def _prepare_input_data_tool(
        self,
        fn: t.Callable,
        name: str,
        args: dict,
        default_data: dict,
    ) -> dict:
        sig = inspect.signature(fn)
        input_data = {}
        for param_name, param in sig.parameters.items():
            if param.kind in (
                inspect.Parameter.VAR_POSITIONAL,
                inspect.Parameter.VAR_KEYWORD,
            ):
                continue

            if param_name in ["state", "config", "tool_call_id"]:
                input_data[param_name] = default_data[param_name]
                continue

            if param_name in INJECTABLE_PARAMS:
                continue

            if (
                hasattr(param, "default")
                and param.default is not inspect._empty
                and hasattr(param.default, "__class__")
            ):
                try:
                    if "Inject" in str(type(param.default)):
                        logger.debug(
                            "Skipping injectable parameter '%s' with Inject syntax",
                            param_name,
                        )
                        continue
                except Exception as exc:  # pragma: no cover - defensive
                    logger.exception("Inject detection failed for '%s': %s", param_name, exc)

            if param_name in args:
                input_data[param_name] = args[param_name]
            elif param.default is inspect.Parameter.empty:
                raise TypeError(f"Missing required parameter '{param_name}' for function '{name}'")

        return input_data

    async def _internal_execute(  # noqa: PLR0915
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        state: AgentState,
        callback_mgr: CallbackManager,
    ) -> Message:
        context = CallbackContext(
            invocation_type=InvocationType.TOOL,
            node_name="ToolNode",
            function_name=name,
            metadata={"tool_call_id": tool_call_id, "args": args, "config": config},
        )

        fn = self._funcs[name]
        input_data = self._prepare_input_data_tool(
            fn,
            name,
            args,
            {
                "tool_call_id": tool_call_id,
                "state": state,
                "config": config,
            },
        )

        meta = {
            "function_name": name,
            "function_argument": args,
            "tool_call_id": tool_call_id,
        }

        event = EventModel.default(
            base_config=config,
            data={
                "tool_call_id": tool_call_id,
                "args": args,
                "function_name": name,
                "is_mcp": False,
            },
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.event_type = EventType.PROGRESS
        event.node_name = "ToolNode"
        event.sequence_id = 1
        publish_event(event)

        def safe_serialize(obj: t.Any) -> dict[str, t.Any]:
            try:
                json.dumps(obj)
                return obj if isinstance(obj, dict) else {"content": obj}
            except (TypeError, OverflowError):
                if hasattr(obj, "model_dump"):
                    dumped = obj.model_dump()  # type: ignore
                    if isinstance(dumped, dict) and dumped.get("type") == "resource":
                        resource = dumped.get("resource", {})
                        if isinstance(resource, dict) and "uri" in resource:
                            resource["uri"] = str(resource["uri"])
                            dumped["resource"] = resource
                    return dumped
                return {"content": str(obj), "type": "fallback"}

        try:
            input_data = await callback_mgr.execute_before_invoke(context, input_data)

            event.event_type = EventType.UPDATE
            event.sequence_id = 2
            event.metadata["status"] = "before_invoke_complete Invoke internal"
            publish_event(event)

            result = await call_sync_or_async(fn, **input_data)

            result = await callback_mgr.execute_after_invoke(
                context,
                input_data,
                result,
            )

            if isinstance(result, Message):
                meta_data = result.metadata or {}
                meta.update(meta_data)
                result.metadata = meta

                event.event_type = EventType.END
                event.data["message"] = result.model_dump()
                event.metadata["status"] = "Internal tool execution complete"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return result

            result_blocks = []
            if isinstance(result, str):
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=result,
                        status="completed",
                        is_error=False,
                    )
                )
            elif isinstance(result, dict):
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=[safe_serialize(result)],
                        status="completed",
                        is_error=False,
                    )
                )
            elif hasattr(result, "model_dump"):
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=[safe_serialize(result.model_dump())],
                        status="completed",
                        is_error=False,
                    )
                )
            elif hasattr(result, "__dict__"):
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=[safe_serialize(result.__dict__)],
                        status="completed",
                        is_error=False,
                    )
                )
            elif isinstance(result, list):
                output = [safe_serialize(item) for item in result]
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=output,
                        status="completed",
                        is_error=False,
                    )
                )
            else:
                result_blocks.append(
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=str(result),
                        status="completed",
                        is_error=False,
                    )
                )

            msg = Message.tool_message(
                content=result_blocks,
                meta=meta,
            )

            event.event_type = EventType.END
            event.data["message"] = msg.model_dump()
            event.metadata["status"] = "Internal tool execution complete"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
            publish_event(event)

            return msg

        except Exception as e:  # pragma: no cover - error path
            recovery_result = await callback_mgr.execute_on_error(context, input_data, e)

            if isinstance(recovery_result, Message):
                event.event_type = EventType.END
                event.data["message"] = recovery_result.model_dump()
                event.metadata["status"] = "Internal tool execution complete, with recovery"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return recovery_result

            event.event_type = EventType.END
            event.data["error"] = str(e)
            event.metadata["status"] = "Internal tool execution complete, with error"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
            publish_event(event)

            return Message.tool_message(
                content=[
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=f"Internal execution error: {e}",
                        status="failed",
                        is_error=True,
                    ),
                    ErrorBlock(message=f"Internal execution error: {e}"),
                ],
                meta=meta,
            )

MCPMixin

Attributes:

Name	Type	Description
mcp_tools	list[str]

Source code in pyagenity/graph/tool_node/executors.py

class MCPMixin:
    _client: t.Any | None
    # The concrete ToolNode defines this
    mcp_tools: list[str]  # type: ignore[assignment]

    def _serialize_result(
        self,
        tool_call_id: str,
        res: t.Any,
    ) -> list[ContentBlock]:
        def safe_serialize(obj: t.Any) -> dict[str, t.Any]:
            try:
                json.dumps(obj)
                return obj if isinstance(obj, dict) else {"content": obj}
            except (TypeError, OverflowError):
                if hasattr(obj, "model_dump"):
                    dumped = obj.model_dump()  # type: ignore
                    if isinstance(dumped, dict) and dumped.get("type") == "resource":
                        resource = dumped.get("resource", {})
                        if isinstance(resource, dict) and "uri" in resource:
                            resource["uri"] = str(resource["uri"])
                            dumped["resource"] = resource
                    return dumped
                return {"content": str(obj), "type": "fallback"}

        for source in [
            getattr(res, "content", None),
            getattr(res, "structured_content", None),
            getattr(res, "data", None),
        ]:
            if source is None:
                continue
            try:
                if isinstance(source, list):
                    result = [safe_serialize(item) for item in source]
                else:
                    result = [safe_serialize(source)]

                return [
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=result,
                        is_error=False,
                        status="completed",
                    )
                ]
            except Exception as e:  # pragma: no cover - defensive
                logger.exception("Serialization failure: %s", e)
                continue

        return [
            ToolResultBlock(
                call_id=tool_call_id,
                output=[
                    {
                        "content": str(res),
                        "type": "fallback",
                    }
                ],
                is_error=False,
                status="completed",
            )
        ]

    async def _get_mcp_tool(self) -> list[dict]:
        tools: list[dict] = []
        if self._client:
            async with self._client:
                res = await self._client.ping()
                if not res:
                    return tools
                mcp_tools: list[t.Any] = await self._client.list_tools()
                for i in mcp_tools:
                    # attribute provided by concrete ToolNode
                    self.mcp_tools.append(i.name)  # type: ignore[attr-defined]
                    tools.append(
                        {
                            "type": "function",
                            "function": {
                                "name": i.name,
                                "description": i.description,
                                "parameters": i.inputSchema,
                            },
                        }
                    )
        return tools

    async def _mcp_execute(
        self,
        name: str,
        args: dict,
        tool_call_id: str,
        config: dict[str, t.Any],
        callback_mgr: CallbackManager,
    ) -> Message:
        context = CallbackContext(
            invocation_type=InvocationType.MCP,
            node_name="ToolNode",
            function_name=name,
            metadata={
                "tool_call_id": tool_call_id,
                "args": args,
                "config": config,
                "mcp_client": bool(self._client),
            },
        )

        meta = {
            "function_name": name,
            "function_argument": args,
            "tool_call_id": tool_call_id,
        }

        event = EventModel.default(
            base_config=config,
            data={
                "tool_call_id": tool_call_id,
                "args": args,
                "function_name": name,
                "is_mcp": True,
            },
            content_type=[ContentType.TOOL_CALL],
            event=Event.TOOL_EXECUTION,
        )
        event.event_type = EventType.PROGRESS
        event.node_name = "ToolNode"
        event.sequence_id = 1
        publish_event(event)

        input_data = {**args}

        try:
            input_data = await callback_mgr.execute_before_invoke(context, input_data)
            event.event_type = EventType.UPDATE
            event.sequence_id = 2
            event.metadata["status"] = "before_invoke_complete Invoke MCP"
            publish_event(event)

            if not self._client:
                error_result = Message.tool_message(
                    content=[
                        ErrorBlock(
                            message="No MCP client configured",
                        ),
                        ToolResultBlock(
                            call_id=tool_call_id,
                            output="No MCP client configured",
                            is_error=True,
                            status="failed",
                        ),
                    ],
                    meta=meta,
                )
                res = await callback_mgr.execute_after_invoke(context, input_data, error_result)
                event.event_type = EventType.ERROR
                event.metadata["error"] = "No MCP client configured"
                publish_event(event)
                return res

            async with self._client:
                if not await self._client.ping():
                    error_result = Message.tool_message(
                        content=[
                            ErrorBlock(message="MCP Server not available. Ping failed."),
                            ToolResultBlock(
                                call_id=tool_call_id,
                                output="MCP Server not available. Ping failed.",
                                is_error=True,
                                status="failed",
                            ),
                        ],
                        meta=meta,
                    )
                    event.event_type = EventType.ERROR
                    event.metadata["error"] = "MCP server not available, ping failed"
                    publish_event(event)
                    return await callback_mgr.execute_after_invoke(
                        context, input_data, error_result
                    )

                res: t.Any = await self._client.call_tool(name, input_data)

                final_res = self._serialize_result(tool_call_id, res)

                result = Message.tool_message(
                    content=final_res,
                    meta=meta,
                )

                res = await callback_mgr.execute_after_invoke(context, input_data, result)
                event.event_type = EventType.END
                event.data["message"] = result.model_dump()
                event.metadata["status"] = "MCP tool execution complete"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return res

        except Exception as e:  # pragma: no cover - error path
            recovery_result = await callback_mgr.execute_on_error(context, input_data, e)

            if isinstance(recovery_result, Message):
                event.event_type = EventType.END
                event.data["message"] = recovery_result.model_dump()
                event.metadata["status"] = "MCP tool execution complete, with recovery"
                event.content_type = [ContentType.TOOL_RESULT, ContentType.MESSAGE]
                publish_event(event)
                return recovery_result

            event.event_type = EventType.END
            event.data["error"] = str(e)
            event.metadata["status"] = "MCP tool execution complete, with recovery"
            event.content_type = [ContentType.TOOL_RESULT, ContentType.ERROR]
            publish_event(event)

            return Message.tool_message(
                content=[
                    ToolResultBlock(
                        call_id=tool_call_id,
                        output=f"MCP execution error: {e}",
                        is_error=True,
                        status="failed",
                    ),
                    ErrorBlock(message=f"MCP execution error: {e}"),
                ],
                meta=meta,
            )

Attributes

mcp_tools instance-attribute

mcp_tools

Functions

schema

Schema utilities and local tool description building for ToolNode.

This module provides the SchemaMixin class which handles automatic schema generation for local Python functions, converting their type annotations and signatures into OpenAI-compatible function schemas. It supports various Python types including primitives, Optional types, List types, and Literal enums.

The schema generation process inspects function signatures and converts them to JSON Schema format suitable for use with language models and function calling APIs.

Classes:

Name	Description
SchemaMixin	Mixin providing schema generation and local tool description building.

Attributes

Classes

SchemaMixin

Mixin providing schema generation and local tool description building.

This mixin provides functionality to automatically generate JSON Schema definitions from Python function signatures. It handles type annotation conversion, parameter analysis, and OpenAI-compatible function schema generation for local tools.

The mixin is designed to be used with ToolNode to automatically generate tool schemas without requiring manual schema definition for Python functions.

Attributes:

Name	Type	Description
_funcs	dict[str, Callable]	Dictionary mapping function names to callable functions. This attribute is expected to be provided by the mixing class.

Methods:

Name	Description
get_local_tool	Generate OpenAI-compatible tool definitions for all registered local functions.

Source code in pyagenity/graph/tool_node/schema.py

class SchemaMixin:
    """Mixin providing schema generation and local tool description building.

    This mixin provides functionality to automatically generate JSON Schema definitions
    from Python function signatures. It handles type annotation conversion, parameter
    analysis, and OpenAI-compatible function schema generation for local tools.

    The mixin is designed to be used with ToolNode to automatically generate tool
    schemas without requiring manual schema definition for Python functions.

    Attributes:
        _funcs: Dictionary mapping function names to callable functions. This
            attribute is expected to be provided by the mixing class.
    """

    _funcs: dict[str, t.Callable]

    @staticmethod
    def _handle_optional_annotation(annotation: t.Any, default: t.Any) -> dict | None:
        """Handle Optional type annotations and convert them to appropriate schemas.

        Processes Optional[T] type annotations (Union[T, None]) and generates
        schema for the non-None type. This method handles the common pattern
        of optional parameters in function signatures.

        Args:
            annotation: The type annotation to process, potentially an Optional type.
            default: The default value for the parameter, used for schema generation.

        Returns:
            Dictionary containing the JSON schema for the non-None type if the
            annotation is Optional, None otherwise.

        Example:
            Optional[str] -> {"type": "string"}
            Optional[int] -> {"type": "integer"}
        """
        args = getattr(annotation, "__args__", None)
        if args and any(a is type(None) for a in args):
            non_none = [a for a in args if a is not type(None)]
            if non_none:
                return SchemaMixin._annotation_to_schema(non_none[0], default)
        return None

    @staticmethod
    def _handle_complex_annotation(annotation: t.Any) -> dict:
        """Handle complex type annotations like List, Literal, and generic types.

        Processes generic type annotations that aren't simple primitive types,
        including container types like List and special types like Literal enums.
        Falls back to string type for unrecognized complex types.

        Args:
            annotation: The complex type annotation to process (e.g., List[str],
                Literal["a", "b", "c"]).

        Returns:
            Dictionary containing the appropriate JSON schema for the complex type.
            For List types, returns array schema with item type.
            For Literal types, returns enum schema with allowed values.
            For unknown types, returns string type as fallback.

        Example:
            List[str] -> {"type": "array", "items": {"type": "string"}}
            Literal["red", "green"] -> {"type": "string", "enum": ["red", "green"]}
        """
        origin = getattr(annotation, "__origin__", None)
        if origin is list:
            item_type = getattr(annotation, "__args__", (str,))[0]
            item_schema = SchemaMixin._annotation_to_schema(item_type, None)
            return {"type": "array", "items": item_schema}

        Literal = getattr(t, "Literal", None)
        if Literal is not None and origin is Literal:
            literals = list(getattr(annotation, "__args__", ()))
            if all(isinstance(literal, str) for literal in literals):
                return {"type": "string", "enum": literals}
            return {"enum": literals}

        return {"type": "string"}

    @staticmethod
    def _annotation_to_schema(annotation: t.Any, default: t.Any) -> dict:
        """Convert a Python type annotation to JSON Schema format.

        Main entry point for type annotation conversion. Handles both simple
        and complex types by delegating to appropriate helper methods.
        Includes default value handling when present.

        Args:
            annotation: The Python type annotation to convert (e.g., str, int,
                Optional[str], List[int]).
            default: The default value for the parameter, included in schema
                if not inspect._empty.

        Returns:
            Dictionary containing the JSON schema representation of the type
            annotation, including default values where applicable.

        Example:
            str -> {"type": "string"}
            int -> {"type": "integer"}
            str with default "hello" -> {"type": "string", "default": "hello"}
        """
        schema = SchemaMixin._handle_optional_annotation(annotation, default)
        if schema:
            return schema

        primitive_mappings = {
            str: {"type": "string"},
            int: {"type": "integer"},
            float: {"type": "number"},
            bool: {"type": "boolean"},
        }

        if annotation in primitive_mappings:
            schema = primitive_mappings[annotation]
        else:
            schema = SchemaMixin._handle_complex_annotation(annotation)

        if default is not inspect._empty:
            schema["default"] = default

        return schema

    def get_local_tool(self) -> list[dict]:
        """Generate OpenAI-compatible tool definitions for all registered local functions.

        Inspects all registered functions in _funcs and automatically generates
        tool schemas by analyzing function signatures, type annotations, and docstrings.
        Excludes injectable parameters that are provided by the framework.

        Returns:
            List of tool definitions in OpenAI function calling format. Each
            definition includes the function name, description (from docstring),
            and complete parameter schema with types and required fields.

        Example:
            For a function:
            ```python
            def calculate(a: int, b: int, operation: str = "add") -> int:
                '''Perform arithmetic calculation.'''
                return a + b if operation == "add" else a - b
            ```

            Returns:
            ```python
            [
                {
                    "type": "function",
                    "function": {
                        "name": "calculate",
                        "description": "Perform arithmetic calculation.",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "a": {"type": "integer"},
                                "b": {"type": "integer"},
                                "operation": {"type": "string", "default": "add"},
                            },
                            "required": ["a", "b"],
                        },
                    },
                }
            ]
            ```

        Note:
            Parameters listed in INJECTABLE_PARAMS (like 'state', 'config',
            'tool_call_id') are automatically excluded from the generated schema
            as they are provided by the framework during execution.
        """
        tools: list[dict] = []
        for name, fn in self._funcs.items():
            sig = inspect.signature(fn)
            params_schema: dict = {"type": "object", "properties": {}, "required": []}

            for p_name, p in sig.parameters.items():
                if p.kind in (
                    inspect.Parameter.VAR_POSITIONAL,
                    inspect.Parameter.VAR_KEYWORD,
                ):
                    continue

                if p_name in INJECTABLE_PARAMS:
                    continue

                annotation = p.annotation if p.annotation is not inspect._empty else str
                prop = SchemaMixin._annotation_to_schema(annotation, p.default)
                params_schema["properties"][p_name] = prop

                if p.default is inspect._empty:
                    params_schema["required"].append(p_name)

            if not params_schema["required"]:
                params_schema.pop("required")

            description = inspect.getdoc(fn) or "No description provided."

            # provider = getattr(fn, "_py_tool_provider", None)
            # tags = getattr(fn, "_py_tool_tags", None)
            # capabilities = getattr(fn, "_py_tool_capabilities", None)

            entry = {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": params_schema,
                },
            }
            # meta: dict[str, t.Any] = {}
            # if provider:
            #     meta["provider"] = provider
            # if tags:
            #     meta["tags"] = tags
            # if capabilities:
            #     meta["capabilities"] = capabilities
            # if meta:
            #     entry["x-pyagenity"] = meta

            tools.append(entry)

        return tools

Functions

get_local_tool

get_local_tool()

Generate OpenAI-compatible tool definitions for all registered local functions.

Inspects all registered functions in _funcs and automatically generates tool schemas by analyzing function signatures, type annotations, and docstrings. Excludes injectable parameters that are provided by the framework.

Returns:

Type	Description
list[dict]	List of tool definitions in OpenAI function calling format. Each
list[dict]	definition includes the function name, description (from docstring),
list[dict]	and complete parameter schema with types and required fields.

Example

For a function:

def calculate(a: int, b: int, operation: str = "add") -> int:
    '''Perform arithmetic calculation.'''
    return a + b if operation == "add" else a - b

Returns:

[
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Perform arithmetic calculation.",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "integer"},
                    "b": {"type": "integer"},
                    "operation": {"type": "string", "default": "add"},
                },
                "required": ["a", "b"],
            },
        },
    }
]

Note

Parameters listed in INJECTABLE_PARAMS (like 'state', 'config', 'tool_call_id') are automatically excluded from the generated schema as they are provided by the framework during execution.

Source code in pyagenity/graph/tool_node/schema.py

def get_local_tool(self) -> list[dict]:
    """Generate OpenAI-compatible tool definitions for all registered local functions.

    Inspects all registered functions in _funcs and automatically generates
    tool schemas by analyzing function signatures, type annotations, and docstrings.
    Excludes injectable parameters that are provided by the framework.

    Returns:
        List of tool definitions in OpenAI function calling format. Each
        definition includes the function name, description (from docstring),
        and complete parameter schema with types and required fields.

    Example:
        For a function:
        ```python
        def calculate(a: int, b: int, operation: str = "add") -> int:
            '''Perform arithmetic calculation.'''
            return a + b if operation == "add" else a - b
        ```

        Returns:
        ```python
        [
            {
                "type": "function",
                "function": {
                    "name": "calculate",
                    "description": "Perform arithmetic calculation.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "a": {"type": "integer"},
                            "b": {"type": "integer"},
                            "operation": {"type": "string", "default": "add"},
                        },
                        "required": ["a", "b"],
                    },
                },
            }
        ]
        ```

    Note:
        Parameters listed in INJECTABLE_PARAMS (like 'state', 'config',
        'tool_call_id') are automatically excluded from the generated schema
        as they are provided by the framework during execution.
    """
    tools: list[dict] = []
    for name, fn in self._funcs.items():
        sig = inspect.signature(fn)
        params_schema: dict = {"type": "object", "properties": {}, "required": []}

        for p_name, p in sig.parameters.items():
            if p.kind in (
                inspect.Parameter.VAR_POSITIONAL,
                inspect.Parameter.VAR_KEYWORD,
            ):
                continue

            if p_name in INJECTABLE_PARAMS:
                continue

            annotation = p.annotation if p.annotation is not inspect._empty else str
            prop = SchemaMixin._annotation_to_schema(annotation, p.default)
            params_schema["properties"][p_name] = prop

            if p.default is inspect._empty:
                params_schema["required"].append(p_name)

        if not params_schema["required"]:
            params_schema.pop("required")

        description = inspect.getdoc(fn) or "No description provided."

        # provider = getattr(fn, "_py_tool_provider", None)
        # tags = getattr(fn, "_py_tool_tags", None)
        # capabilities = getattr(fn, "_py_tool_capabilities", None)

        entry = {
            "type": "function",
            "function": {
                "name": name,
                "description": description,
                "parameters": params_schema,
            },
        }
        # meta: dict[str, t.Any] = {}
        # if provider:
        #     meta["provider"] = provider
        # if tags:
        #     meta["tags"] = tags
        # if capabilities:
        #     meta["capabilities"] = capabilities
        # if meta:
        #     entry["x-pyagenity"] = meta

        tools.append(entry)

    return tools