Tools

Integration adapters for optional third-party SDKs.

This module exposes unified wrappers for integrating external tool registries and SDKs with PyAgenity agent graphs. The adapters provide registry-based discovery, function-calling schemas, and normalized execution for supported tool providers.

Exports

ComposioAdapter: Adapter for the Composio Python SDK. LangChainAdapter: Adapter for LangChain tool registry and execution.

Modules:

Name	Description
composio_adapter	Composio adapter for PyAgenity.
langchain_adapter	LangChain adapter for PyAgenity (generic wrapper, registry-based).

Classes:

Name	Description
ComposioAdapter	Adapter around Composio Python SDK.
LangChainAdapter	Generic registry-based LangChain adapter.

Attributes

__all__ module-attribute

__all__ = ['ComposioAdapter', 'LangChainAdapter']

Classes

ComposioAdapter

Adapter around Composio Python SDK.

Notes on SDK methods used (from docs): - composio.tools.get(user_id=..., tools=[...]/toolkits=[...]/search=..., scopes=..., limit=...) Returns tools formatted for providers or agent frameworks; includes schema. - composio.tools.get_raw_composio_tools(...) Returns raw tool schemas including input_parameters. - composio.tools.execute(slug, arguments, user_id=..., connected_account_id=..., ...) Executes a tool and returns a dict like {data, successful, error}.

Methods:

Name	Description
__init__	Initialize the ComposioAdapter.
execute	Execute a Composio tool and return a normalized response dict.
is_available	Return True if composio SDK is importable.
list_raw_tools_for_llm	Return raw Composio tool schemas mapped to function-calling format.
list_tools_for_llm	Return tools formatted for LLM function-calling.

Source code in pyagenity/adapters/tools/composio_adapter.py

class ComposioAdapter:
    """Adapter around Composio Python SDK.

    Notes on SDK methods used (from docs):
    - composio.tools.get(user_id=..., tools=[...]/toolkits=[...]/search=..., scopes=..., limit=...)
        Returns tools formatted for providers or agent frameworks; includes schema.
    - composio.tools.get_raw_composio_tools(...)
        Returns raw tool schemas including input_parameters.
    - composio.tools.execute(slug, arguments, user_id=..., connected_account_id=..., ...)
        Executes a tool and returns a dict like {data, successful, error}.
    """

    def __init__(
        self,
        *,
        api_key: str | None = None,
        provider: t.Any | None = None,
        file_download_dir: str | None = None,
        toolkit_versions: t.Any | None = None,
    ) -> None:
        """
        Initialize the ComposioAdapter.

        Args:
            api_key (str | None): Optional API key for Composio.
            provider (Any | None): Optional provider integration.
            file_download_dir (str | None): Directory for auto file handling.
            toolkit_versions (Any | None): Toolkit version overrides.

        Raises:
            ImportError: If composio SDK is not installed.
        """
        if not HAS_COMPOSIO:
            raise ImportError(
                "ComposioAdapter requires 'composio' package. Install with: "
                "pip install pyagenity[composio]"
            )

        self._composio = Composio(  # type: ignore[call-arg]
            api_key=api_key,
            provider=provider,
            file_download_dir=file_download_dir,
            toolkit_versions=toolkit_versions,
        )

    @staticmethod
    def is_available() -> bool:
        """
        Return True if composio SDK is importable.

        Returns:
            bool: True if composio SDK is available, False otherwise.
        """
        return HAS_COMPOSIO

    def list_tools_for_llm(
        self,
        *,
        user_id: str,
        tool_slugs: list[str] | None = None,
        toolkits: list[str] | None = None,
        search: str | None = None,
        scopes: list[str] | None = None,
        limit: int | None = None,
    ) -> list[dict[str, t.Any]]:
        """
        Return tools formatted for LLM function-calling.

        Args:
            user_id (str): User ID for tool discovery.
            tool_slugs (list[str] | None): Optional list of tool slugs.
            toolkits (list[str] | None): Optional list of toolkits.
            search (str | None): Optional search string.
            scopes (list[str] | None): Optional scopes.
            limit (int | None): Optional limit on number of tools.

        Returns:
            list[dict[str, Any]]: List of tools in function-calling format.
        """
        # Prefer the provider-wrapped format when available
        tools = self._composio.tools.get(
            user_id=user_id,
            tools=tool_slugs,  # type: ignore[arg-type]
            toolkits=toolkits,  # type: ignore[arg-type]
            search=search,
            scopes=scopes,
            limit=limit,
        )

        # The provider-wrapped output may already be in the desired structure.
        # We'll detect and pass-through; otherwise convert using raw schemas.
        formatted: list[dict[str, t.Any]] = []
        for t_obj in tools if isinstance(tools, list) else []:
            try:
                if (
                    isinstance(t_obj, dict)
                    and t_obj.get("type") == "function"
                    and "function" in t_obj
                ):
                    formatted.append(t_obj)
                else:
                    # Fallback: try to pull minimal fields
                    fn = t_obj.get("function", {}) if isinstance(t_obj, dict) else {}
                    if fn.get("name") and fn.get("parameters"):
                        formatted.append({"type": "function", "function": fn})
            except Exception as exc:
                logger.debug("Skipping non-conforming Composio tool wrapper: %s", exc)
                continue

        if formatted:
            return formatted

        # Fallback to raw schemas and convert manually
        formatted.extend(
            self.list_raw_tools_for_llm(
                tool_slugs=tool_slugs, toolkits=toolkits, search=search, scopes=scopes, limit=limit
            )
        )

        return formatted

    def list_raw_tools_for_llm(
        self,
        *,
        tool_slugs: list[str] | None = None,
        toolkits: list[str] | None = None,
        search: str | None = None,
        scopes: list[str] | None = None,
        limit: int | None = None,
    ) -> list[dict[str, t.Any]]:
        """
        Return raw Composio tool schemas mapped to function-calling format.

        Args:
            tool_slugs (list[str] | None): Optional list of tool slugs.
            toolkits (list[str] | None): Optional list of toolkits.
            search (str | None): Optional search string.
            scopes (list[str] | None): Optional scopes.
            limit (int | None): Optional limit on number of tools.

        Returns:
            list[dict[str, Any]]: List of raw tool schemas in function-calling format.
        """
        formatted: list[dict[str, t.Any]] = []
        raw_tools = self._composio.tools.get_raw_composio_tools(
            tools=tool_slugs, search=search, toolkits=toolkits, scopes=scopes, limit=limit
        )

        for tool in raw_tools:
            try:
                name = tool.slug  # type: ignore[attr-defined]
                description = getattr(tool, "description", "") or "Composio tool"
                params = getattr(tool, "input_parameters", None)
                if not params:
                    # Minimal shape if schema missing
                    params = {"type": "object", "properties": {}}
                formatted.append(
                    {
                        "type": "function",
                        "function": {
                            "name": name,
                            "description": description,
                            "parameters": params,
                        },
                    }
                )
            except Exception as e:
                logger.warning("Failed to map Composio tool schema: %s", e)
                continue
        return formatted

    def execute(
        self,
        *,
        slug: str,
        arguments: dict[str, t.Any],
        user_id: str | None = None,
        connected_account_id: str | None = None,
        custom_auth_params: dict[str, t.Any] | None = None,
        custom_connection_data: dict[str, t.Any] | None = None,
        text: str | None = None,
        version: str | None = None,
        toolkit_versions: t.Any | None = None,
        modifiers: t.Any | None = None,
    ) -> dict[str, t.Any]:
        """
        Execute a Composio tool and return a normalized response dict.

        Args:
            slug (str): Tool slug to execute.
            arguments (dict[str, Any]): Arguments for the tool.
            user_id (str | None): Optional user ID.
            connected_account_id (str | None): Optional connected account ID.
            custom_auth_params (dict[str, Any] | None): Optional custom auth params.
            custom_connection_data (dict[str, Any] | None): Optional custom connection data.
            text (str | None): Optional text input.
            version (str | None): Optional version.
            toolkit_versions (Any | None): Optional toolkit versions.
            modifiers (Any | None): Optional modifiers.

        Returns:
            dict[str, Any]: Normalized response dict with keys: successful, data, error.
        """
        resp = self._composio.tools.execute(
            slug=slug,
            arguments=arguments,
            user_id=user_id,
            connected_account_id=connected_account_id,
            custom_auth_params=custom_auth_params,
            custom_connection_data=custom_connection_data,
            text=text,
            version=version,
            toolkit_versions=toolkit_versions,
            modifiers=modifiers,
        )

        # The SDK returns a TypedDict-like object; ensure plain dict
        if hasattr(resp, "copy") and not isinstance(resp, dict):  # e.g., TypedDict proxy
            try:
                resp = dict(resp)  # type: ignore[assignment]
            except Exception as exc:
                logger.debug("Could not coerce Composio response to dict: %s", exc)

        # Normalize key presence
        successful = bool(resp.get("successful", False))  # type: ignore[arg-type]
        data = resp.get("data")
        error = resp.get("error")
        return {"successful": successful, "data": data, "error": error}

Functions

__init__

__init__(*, api_key=None, provider=None, file_download_dir=None, toolkit_versions=None)

Initialize the ComposioAdapter.

Parameters:

Name	Type	Description	Default
api_key	str | None	Optional API key for Composio.	None
provider	Any | None	Optional provider integration.	None
file_download_dir	str | None	Directory for auto file handling.	None
toolkit_versions	Any | None	Toolkit version overrides.	None

Raises:

Type	Description
ImportError	If composio SDK is not installed.

Source code in pyagenity/adapters/tools/composio_adapter.py

def __init__(
    self,
    *,
    api_key: str | None = None,
    provider: t.Any | None = None,
    file_download_dir: str | None = None,
    toolkit_versions: t.Any | None = None,
) -> None:
    """
    Initialize the ComposioAdapter.

    Args:
        api_key (str | None): Optional API key for Composio.
        provider (Any | None): Optional provider integration.
        file_download_dir (str | None): Directory for auto file handling.
        toolkit_versions (Any | None): Toolkit version overrides.

    Raises:
        ImportError: If composio SDK is not installed.
    """
    if not HAS_COMPOSIO:
        raise ImportError(
            "ComposioAdapter requires 'composio' package. Install with: "
            "pip install pyagenity[composio]"
        )

    self._composio = Composio(  # type: ignore[call-arg]
        api_key=api_key,
        provider=provider,
        file_download_dir=file_download_dir,
        toolkit_versions=toolkit_versions,
    )

execute

execute(*, slug, arguments, user_id=None, connected_account_id=None, custom_auth_params=None, custom_connection_data=None, text=None, version=None, toolkit_versions=None, modifiers=None)

Execute a Composio tool and return a normalized response dict.

Parameters:

Name	Type	Description	Default
slug	str	Tool slug to execute.	required
arguments	dict[str, Any]	Arguments for the tool.	required
user_id	str | None	Optional user ID.	None
connected_account_id	str | None	Optional connected account ID.	None
custom_auth_params	dict[str, Any] | None	Optional custom auth params.	None
custom_connection_data	dict[str, Any] | None	Optional custom connection data.	None
text	str | None	Optional text input.	None
version	str | None	Optional version.	None
toolkit_versions	Any | None	Optional toolkit versions.	None
modifiers	Any | None	Optional modifiers.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Normalized response dict with keys: successful, data, error.

Source code in pyagenity/adapters/tools/composio_adapter.py

def execute(
    self,
    *,
    slug: str,
    arguments: dict[str, t.Any],
    user_id: str | None = None,
    connected_account_id: str | None = None,
    custom_auth_params: dict[str, t.Any] | None = None,
    custom_connection_data: dict[str, t.Any] | None = None,
    text: str | None = None,
    version: str | None = None,
    toolkit_versions: t.Any | None = None,
    modifiers: t.Any | None = None,
) -> dict[str, t.Any]:
    """
    Execute a Composio tool and return a normalized response dict.

    Args:
        slug (str): Tool slug to execute.
        arguments (dict[str, Any]): Arguments for the tool.
        user_id (str | None): Optional user ID.
        connected_account_id (str | None): Optional connected account ID.
        custom_auth_params (dict[str, Any] | None): Optional custom auth params.
        custom_connection_data (dict[str, Any] | None): Optional custom connection data.
        text (str | None): Optional text input.
        version (str | None): Optional version.
        toolkit_versions (Any | None): Optional toolkit versions.
        modifiers (Any | None): Optional modifiers.

    Returns:
        dict[str, Any]: Normalized response dict with keys: successful, data, error.
    """
    resp = self._composio.tools.execute(
        slug=slug,
        arguments=arguments,
        user_id=user_id,
        connected_account_id=connected_account_id,
        custom_auth_params=custom_auth_params,
        custom_connection_data=custom_connection_data,
        text=text,
        version=version,
        toolkit_versions=toolkit_versions,
        modifiers=modifiers,
    )

    # The SDK returns a TypedDict-like object; ensure plain dict
    if hasattr(resp, "copy") and not isinstance(resp, dict):  # e.g., TypedDict proxy
        try:
            resp = dict(resp)  # type: ignore[assignment]
        except Exception as exc:
            logger.debug("Could not coerce Composio response to dict: %s", exc)

    # Normalize key presence
    successful = bool(resp.get("successful", False))  # type: ignore[arg-type]
    data = resp.get("data")
    error = resp.get("error")
    return {"successful": successful, "data": data, "error": error}

is_available staticmethod

is_available()

Return True if composio SDK is importable.

Returns:

Name	Type	Description
bool	bool	True if composio SDK is available, False otherwise.

Source code in pyagenity/adapters/tools/composio_adapter.py

@staticmethod
def is_available() -> bool:
    """
    Return True if composio SDK is importable.

    Returns:
        bool: True if composio SDK is available, False otherwise.
    """
    return HAS_COMPOSIO

list_raw_tools_for_llm

list_raw_tools_for_llm(*, tool_slugs=None, toolkits=None, search=None, scopes=None, limit=None)

Return raw Composio tool schemas mapped to function-calling format.

Parameters:

Name	Type	Description	Default
tool_slugs	list[str] | None	Optional list of tool slugs.	None
toolkits	list[str] | None	Optional list of toolkits.	None
search	str | None	Optional search string.	None
scopes	list[str] | None	Optional scopes.	None
limit	int | None	Optional limit on number of tools.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of raw tool schemas in function-calling format.

Source code in pyagenity/adapters/tools/composio_adapter.py

def list_raw_tools_for_llm(
    self,
    *,
    tool_slugs: list[str] | None = None,
    toolkits: list[str] | None = None,
    search: str | None = None,
    scopes: list[str] | None = None,
    limit: int | None = None,
) -> list[dict[str, t.Any]]:
    """
    Return raw Composio tool schemas mapped to function-calling format.

    Args:
        tool_slugs (list[str] | None): Optional list of tool slugs.
        toolkits (list[str] | None): Optional list of toolkits.
        search (str | None): Optional search string.
        scopes (list[str] | None): Optional scopes.
        limit (int | None): Optional limit on number of tools.

    Returns:
        list[dict[str, Any]]: List of raw tool schemas in function-calling format.
    """
    formatted: list[dict[str, t.Any]] = []
    raw_tools = self._composio.tools.get_raw_composio_tools(
        tools=tool_slugs, search=search, toolkits=toolkits, scopes=scopes, limit=limit
    )

    for tool in raw_tools:
        try:
            name = tool.slug  # type: ignore[attr-defined]
            description = getattr(tool, "description", "") or "Composio tool"
            params = getattr(tool, "input_parameters", None)
            if not params:
                # Minimal shape if schema missing
                params = {"type": "object", "properties": {}}
            formatted.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": description,
                        "parameters": params,
                    },
                }
            )
        except Exception as e:
            logger.warning("Failed to map Composio tool schema: %s", e)
            continue
    return formatted

list_tools_for_llm

list_tools_for_llm(*, user_id, tool_slugs=None, toolkits=None, search=None, scopes=None, limit=None)

Return tools formatted for LLM function-calling.

Parameters:

Name	Type	Description	Default
user_id	str	User ID for tool discovery.	required
tool_slugs	list[str] | None	Optional list of tool slugs.	None
toolkits	list[str] | None	Optional list of toolkits.	None
search	str | None	Optional search string.	None
scopes	list[str] | None	Optional scopes.	None
limit	int | None	Optional limit on number of tools.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of tools in function-calling format.

Source code in pyagenity/adapters/tools/composio_adapter.py

def list_tools_for_llm(
    self,
    *,
    user_id: str,
    tool_slugs: list[str] | None = None,
    toolkits: list[str] | None = None,
    search: str | None = None,
    scopes: list[str] | None = None,
    limit: int | None = None,
) -> list[dict[str, t.Any]]:
    """
    Return tools formatted for LLM function-calling.

    Args:
        user_id (str): User ID for tool discovery.
        tool_slugs (list[str] | None): Optional list of tool slugs.
        toolkits (list[str] | None): Optional list of toolkits.
        search (str | None): Optional search string.
        scopes (list[str] | None): Optional scopes.
        limit (int | None): Optional limit on number of tools.

    Returns:
        list[dict[str, Any]]: List of tools in function-calling format.
    """
    # Prefer the provider-wrapped format when available
    tools = self._composio.tools.get(
        user_id=user_id,
        tools=tool_slugs,  # type: ignore[arg-type]
        toolkits=toolkits,  # type: ignore[arg-type]
        search=search,
        scopes=scopes,
        limit=limit,
    )

    # The provider-wrapped output may already be in the desired structure.
    # We'll detect and pass-through; otherwise convert using raw schemas.
    formatted: list[dict[str, t.Any]] = []
    for t_obj in tools if isinstance(tools, list) else []:
        try:
            if (
                isinstance(t_obj, dict)
                and t_obj.get("type") == "function"
                and "function" in t_obj
            ):
                formatted.append(t_obj)
            else:
                # Fallback: try to pull minimal fields
                fn = t_obj.get("function", {}) if isinstance(t_obj, dict) else {}
                if fn.get("name") and fn.get("parameters"):
                    formatted.append({"type": "function", "function": fn})
        except Exception as exc:
            logger.debug("Skipping non-conforming Composio tool wrapper: %s", exc)
            continue

    if formatted:
        return formatted

    # Fallback to raw schemas and convert manually
    formatted.extend(
        self.list_raw_tools_for_llm(
            tool_slugs=tool_slugs, toolkits=toolkits, search=search, scopes=scopes, limit=limit
        )
    )

    return formatted

LangChainAdapter

Generic registry-based LangChain adapter.

Notes

• Avoids importing heavy integrations until needed (lazy default autoload).
• Normalizes schemas and execution results into simple dicts.
• Allows arbitrary tool registration instead of hardcoding a tiny set.

Methods:

Name	Description
__init__	Initialize LangChainAdapter.
execute	Execute a supported LangChain tool and normalize the response.
is_available	Return True if langchain-core is importable.
list_tools_for_llm	Return a list of function-calling formatted tool schemas.
register_tool	Register a tool instance and return the resolved name used for exposure.
register_tools	Register multiple tool instances.

Source code in pyagenity/adapters/tools/langchain_adapter.py

class LangChainAdapter:
    """
    Generic registry-based LangChain adapter.

    Notes:
        - Avoids importing heavy integrations until needed (lazy default autoload).
        - Normalizes schemas and execution results into simple dicts.
        - Allows arbitrary tool registration instead of hardcoding a tiny set.
    """

    def __init__(self, *, autoload_default_tools: bool = True) -> None:
        """
        Initialize LangChainAdapter.

        Args:
            autoload_default_tools (bool): Whether to autoload default tools if registry is empty.

        Raises:
            ImportError: If langchain-core is not installed.
        """
        if not HAS_LANGCHAIN:
            raise ImportError(
                "LangChainAdapter requires 'langchain-core' and optional integrations.\n"
                "Install with: pip install pyagenity[langchain]"
            )
        self._registry: dict[str, LangChainToolWrapper] = {}
        self._autoload = autoload_default_tools

    @staticmethod
    def is_available() -> bool:
        """
        Return True if langchain-core is importable.

        Returns:
            bool: True if langchain-core is available, False otherwise.
        """
        return HAS_LANGCHAIN

    # ------------------------
    # Discovery
    # ------------------------
    def list_tools_for_llm(self) -> list[dict[str, t.Any]]:
        """
        Return a list of function-calling formatted tool schemas.

        If registry is empty and autoload is enabled, attempt to autoload a
        couple of common tools for convenience (tavily_search, requests_get).

        Returns:
            list[dict[str, Any]]: List of tool schemas in function-calling format.
        """
        if not self._registry and self._autoload:
            self._try_autoload_defaults()

        return [wrapper.to_schema() for wrapper in self._registry.values()]

    # ------------------------
    # Execute
    # ------------------------
    def execute(self, *, name: str, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
        """
        Execute a supported LangChain tool and normalize the response.

        Args:
            name (str): Name of the tool to execute.
            arguments (dict[str, Any]): Arguments for the tool.

        Returns:
            dict[str, Any]: Normalized response dict with keys: successful, data, error.
        """
        if name not in self._registry and self._autoload:
            # Late autoload attempt in case discovery wasn't called first
            self._try_autoload_defaults()

        wrapper = self._registry.get(name)
        if not wrapper:
            return {"successful": False, "data": None, "error": f"Unknown LangChain tool: {name}"}
        return wrapper.execute(arguments)

    # ------------------------
    # Internals
    # ------------------------
    def register_tool(
        self,
        tool: t.Any,
        *,
        name: str | None = None,
        description: str | None = None,
    ) -> str:
        """
        Register a tool instance and return the resolved name used for exposure.

        Args:
            tool (Any): Tool instance to register.
            name (str | None): Optional override for tool name.
            description (str | None): Optional override for tool description.

        Returns:
            str: The resolved name used for exposure.
        """
        wrapper = LangChainToolWrapper(tool, name=name, description=description)
        self._registry[wrapper.name] = wrapper
        return wrapper.name

    def register_tools(self, tools: list[t.Any]) -> list[str]:
        """
        Register multiple tool instances.

        Args:
            tools (list[Any]): List of tool instances to register.

        Returns:
            list[str]: List of resolved names for the registered tools.
        """
        names: list[str] = []
        for tool in tools:
            names.append(self.register_tool(tool))
        return names

    def _create_tavily_search_tool(self) -> t.Any:
        """
        Construct Tavily search tool lazily.

        Prefer the new dedicated integration `langchain_tavily.TavilySearch`.
        Fall back to the deprecated community tool if needed.

        Returns:
            Any: Tavily search tool instance.

        Raises:
            ImportError: If Tavily tool cannot be imported.
        """
        # Preferred: langchain-tavily
        try:
            mod = importlib.import_module("langchain_tavily")
            return mod.TavilySearch()  # type: ignore[attr-defined]
        except Exception as exc:
            logger.debug("Preferred langchain_tavily import failed: %s", exc)

        # Fallback: deprecated community tool (still functional for now)
        try:
            mod = importlib.import_module("langchain_community.tools.tavily_search")
            return mod.TavilySearchResults()
        except Exception as exc:  # ImportError or runtime
            raise ImportError(
                "Tavily tool requires 'langchain-tavily' (preferred) or"
                " 'langchain-community' with 'tavily-python'.\n"
                "Install with: pip install pyagenity[langchain]"
            ) from exc

    def _create_requests_get_tool(self) -> t.Any:
        """
        Construct RequestsGetTool lazily with a basic requests wrapper.

        Note: Requests tools require an explicit wrapper instance and, for safety,
        default to disallowing dangerous requests. Here we opt-in to allow GET
        requests by setting allow_dangerous_requests=True to make the tool usable
        in agent contexts. Consider tightening this in your application.

        Returns:
            Any: RequestsGetTool instance.

        Raises:
            ImportError: If RequestsGetTool cannot be imported.
        """
        try:
            req_tool_mod = importlib.import_module("langchain_community.tools.requests.tool")
            util_mod = importlib.import_module("langchain_community.utilities.requests")
            wrapper = util_mod.TextRequestsWrapper(headers={})  # type: ignore[attr-defined]
            return req_tool_mod.RequestsGetTool(
                requests_wrapper=wrapper,
                allow_dangerous_requests=True,
            )
        except Exception as exc:  # ImportError or runtime
            raise ImportError(
                "Requests tool requires 'langchain-community'.\n"
                "Install with: pip install pyagenity[langchain]"
            ) from exc

    def _try_autoload_defaults(self) -> None:
        """
        Best-effort autoload of a couple of common tools.

        This keeps prior behavior available while allowing users to register
        arbitrary tools. Failures are logged but non-fatal.

        Returns:
            None
        """
        # Tavily search
        try:
            tavily = self._create_tavily_search_tool()
            self.register_tool(tavily, name="tavily_search")
        except Exception as exc:
            logger.debug("Skipping Tavily autoload: %s", exc)

        # Requests GET
        try:
            rget = self._create_requests_get_tool()
            self.register_tool(rget, name="requests_get")
        except Exception as exc:
            logger.debug("Skipping requests_get autoload: %s", exc)

Functions

__init__

__init__(*, autoload_default_tools=True)

Initialize LangChainAdapter.

Parameters:

Name	Type	Description	Default
autoload_default_tools	bool	Whether to autoload default tools if registry is empty.	True

Raises:

Type	Description
ImportError	If langchain-core is not installed.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def __init__(self, *, autoload_default_tools: bool = True) -> None:
    """
    Initialize LangChainAdapter.

    Args:
        autoload_default_tools (bool): Whether to autoload default tools if registry is empty.

    Raises:
        ImportError: If langchain-core is not installed.
    """
    if not HAS_LANGCHAIN:
        raise ImportError(
            "LangChainAdapter requires 'langchain-core' and optional integrations.\n"
            "Install with: pip install pyagenity[langchain]"
        )
    self._registry: dict[str, LangChainToolWrapper] = {}
    self._autoload = autoload_default_tools

execute

execute(*, name, arguments)

Execute a supported LangChain tool and normalize the response.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool to execute.	required
arguments	dict[str, Any]	Arguments for the tool.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Normalized response dict with keys: successful, data, error.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def execute(self, *, name: str, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
    """
    Execute a supported LangChain tool and normalize the response.

    Args:
        name (str): Name of the tool to execute.
        arguments (dict[str, Any]): Arguments for the tool.

    Returns:
        dict[str, Any]: Normalized response dict with keys: successful, data, error.
    """
    if name not in self._registry and self._autoload:
        # Late autoload attempt in case discovery wasn't called first
        self._try_autoload_defaults()

    wrapper = self._registry.get(name)
    if not wrapper:
        return {"successful": False, "data": None, "error": f"Unknown LangChain tool: {name}"}
    return wrapper.execute(arguments)

is_available staticmethod

is_available()

Return True if langchain-core is importable.

Returns:

Name	Type	Description
bool	bool	True if langchain-core is available, False otherwise.

Source code in pyagenity/adapters/tools/langchain_adapter.py

@staticmethod
def is_available() -> bool:
    """
    Return True if langchain-core is importable.

    Returns:
        bool: True if langchain-core is available, False otherwise.
    """
    return HAS_LANGCHAIN

list_tools_for_llm

list_tools_for_llm()

Return a list of function-calling formatted tool schemas.

If registry is empty and autoload is enabled, attempt to autoload a couple of common tools for convenience (tavily_search, requests_get).

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of tool schemas in function-calling format.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def list_tools_for_llm(self) -> list[dict[str, t.Any]]:
    """
    Return a list of function-calling formatted tool schemas.

    If registry is empty and autoload is enabled, attempt to autoload a
    couple of common tools for convenience (tavily_search, requests_get).

    Returns:
        list[dict[str, Any]]: List of tool schemas in function-calling format.
    """
    if not self._registry and self._autoload:
        self._try_autoload_defaults()

    return [wrapper.to_schema() for wrapper in self._registry.values()]

register_tool

register_tool(tool, *, name=None, description=None)

Register a tool instance and return the resolved name used for exposure.

Parameters:

Name	Type	Description	Default
tool	Any	Tool instance to register.	required
name	str | None	Optional override for tool name.	None
description	str | None	Optional override for tool description.	None

Returns:

Name	Type	Description
str	str	The resolved name used for exposure.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def register_tool(
    self,
    tool: t.Any,
    *,
    name: str | None = None,
    description: str | None = None,
) -> str:
    """
    Register a tool instance and return the resolved name used for exposure.

    Args:
        tool (Any): Tool instance to register.
        name (str | None): Optional override for tool name.
        description (str | None): Optional override for tool description.

    Returns:
        str: The resolved name used for exposure.
    """
    wrapper = LangChainToolWrapper(tool, name=name, description=description)
    self._registry[wrapper.name] = wrapper
    return wrapper.name

register_tools

register_tools(tools)

Register multiple tool instances.

Parameters:

Name	Type	Description	Default
tools	list[Any]	List of tool instances to register.	required

Returns:

Type	Description
list[str]	list[str]: List of resolved names for the registered tools.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def register_tools(self, tools: list[t.Any]) -> list[str]:
    """
    Register multiple tool instances.

    Args:
        tools (list[Any]): List of tool instances to register.

    Returns:
        list[str]: List of resolved names for the registered tools.
    """
    names: list[str] = []
    for tool in tools:
        names.append(self.register_tool(tool))
    return names

Modules

composio_adapter

Composio adapter for PyAgenity.

This module provides a thin wrapper around the Composio Python SDK to: - Fetch tools formatted for LLM function calling (matching ToolNode format) - Execute Composio tools directly

The dependency is optional. Install with: pip install pyagenity[composio]

Usage outline

adapter = ComposioAdapter(api_key=os.environ["COMPOSIO_API_KEY"]) # optional key result = adapter.execute( slug="GITHUB_LIST_STARGAZERS", arguments={"owner": "ComposioHQ", "repo": "composio"}, user_id="user-123", )

Classes:

Name	Description
ComposioAdapter	Adapter around Composio Python SDK.

Attributes:

Name	Type	Description
HAS_COMPOSIO
logger

Attributes

HAS_COMPOSIO module-attribute

HAS_COMPOSIO = True

logger module-attribute

logger = getLogger(__name__)

Classes

ComposioAdapter

Adapter around Composio Python SDK.

Notes on SDK methods used (from docs): - composio.tools.get(user_id=..., tools=[...]/toolkits=[...]/search=..., scopes=..., limit=...) Returns tools formatted for providers or agent frameworks; includes schema. - composio.tools.get_raw_composio_tools(...) Returns raw tool schemas including input_parameters. - composio.tools.execute(slug, arguments, user_id=..., connected_account_id=..., ...) Executes a tool and returns a dict like {data, successful, error}.

Methods:

Name	Description
__init__	Initialize the ComposioAdapter.
execute	Execute a Composio tool and return a normalized response dict.
is_available	Return True if composio SDK is importable.
list_raw_tools_for_llm	Return raw Composio tool schemas mapped to function-calling format.
list_tools_for_llm	Return tools formatted for LLM function-calling.

Source code in pyagenity/adapters/tools/composio_adapter.py

class ComposioAdapter:
    """Adapter around Composio Python SDK.

    Notes on SDK methods used (from docs):
    - composio.tools.get(user_id=..., tools=[...]/toolkits=[...]/search=..., scopes=..., limit=...)
        Returns tools formatted for providers or agent frameworks; includes schema.
    - composio.tools.get_raw_composio_tools(...)
        Returns raw tool schemas including input_parameters.
    - composio.tools.execute(slug, arguments, user_id=..., connected_account_id=..., ...)
        Executes a tool and returns a dict like {data, successful, error}.
    """

    def __init__(
        self,
        *,
        api_key: str | None = None,
        provider: t.Any | None = None,
        file_download_dir: str | None = None,
        toolkit_versions: t.Any | None = None,
    ) -> None:
        """
        Initialize the ComposioAdapter.

        Args:
            api_key (str | None): Optional API key for Composio.
            provider (Any | None): Optional provider integration.
            file_download_dir (str | None): Directory for auto file handling.
            toolkit_versions (Any | None): Toolkit version overrides.

        Raises:
            ImportError: If composio SDK is not installed.
        """
        if not HAS_COMPOSIO:
            raise ImportError(
                "ComposioAdapter requires 'composio' package. Install with: "
                "pip install pyagenity[composio]"
            )

        self._composio = Composio(  # type: ignore[call-arg]
            api_key=api_key,
            provider=provider,
            file_download_dir=file_download_dir,
            toolkit_versions=toolkit_versions,
        )

    @staticmethod
    def is_available() -> bool:
        """
        Return True if composio SDK is importable.

        Returns:
            bool: True if composio SDK is available, False otherwise.
        """
        return HAS_COMPOSIO

    def list_tools_for_llm(
        self,
        *,
        user_id: str,
        tool_slugs: list[str] | None = None,
        toolkits: list[str] | None = None,
        search: str | None = None,
        scopes: list[str] | None = None,
        limit: int | None = None,
    ) -> list[dict[str, t.Any]]:
        """
        Return tools formatted for LLM function-calling.

        Args:
            user_id (str): User ID for tool discovery.
            tool_slugs (list[str] | None): Optional list of tool slugs.
            toolkits (list[str] | None): Optional list of toolkits.
            search (str | None): Optional search string.
            scopes (list[str] | None): Optional scopes.
            limit (int | None): Optional limit on number of tools.

        Returns:
            list[dict[str, Any]]: List of tools in function-calling format.
        """
        # Prefer the provider-wrapped format when available
        tools = self._composio.tools.get(
            user_id=user_id,
            tools=tool_slugs,  # type: ignore[arg-type]
            toolkits=toolkits,  # type: ignore[arg-type]
            search=search,
            scopes=scopes,
            limit=limit,
        )

        # The provider-wrapped output may already be in the desired structure.
        # We'll detect and pass-through; otherwise convert using raw schemas.
        formatted: list[dict[str, t.Any]] = []
        for t_obj in tools if isinstance(tools, list) else []:
            try:
                if (
                    isinstance(t_obj, dict)
                    and t_obj.get("type") == "function"
                    and "function" in t_obj
                ):
                    formatted.append(t_obj)
                else:
                    # Fallback: try to pull minimal fields
                    fn = t_obj.get("function", {}) if isinstance(t_obj, dict) else {}
                    if fn.get("name") and fn.get("parameters"):
                        formatted.append({"type": "function", "function": fn})
            except Exception as exc:
                logger.debug("Skipping non-conforming Composio tool wrapper: %s", exc)
                continue

        if formatted:
            return formatted

        # Fallback to raw schemas and convert manually
        formatted.extend(
            self.list_raw_tools_for_llm(
                tool_slugs=tool_slugs, toolkits=toolkits, search=search, scopes=scopes, limit=limit
            )
        )

        return formatted

    def list_raw_tools_for_llm(
        self,
        *,
        tool_slugs: list[str] | None = None,
        toolkits: list[str] | None = None,
        search: str | None = None,
        scopes: list[str] | None = None,
        limit: int | None = None,
    ) -> list[dict[str, t.Any]]:
        """
        Return raw Composio tool schemas mapped to function-calling format.

        Args:
            tool_slugs (list[str] | None): Optional list of tool slugs.
            toolkits (list[str] | None): Optional list of toolkits.
            search (str | None): Optional search string.
            scopes (list[str] | None): Optional scopes.
            limit (int | None): Optional limit on number of tools.

        Returns:
            list[dict[str, Any]]: List of raw tool schemas in function-calling format.
        """
        formatted: list[dict[str, t.Any]] = []
        raw_tools = self._composio.tools.get_raw_composio_tools(
            tools=tool_slugs, search=search, toolkits=toolkits, scopes=scopes, limit=limit
        )

        for tool in raw_tools:
            try:
                name = tool.slug  # type: ignore[attr-defined]
                description = getattr(tool, "description", "") or "Composio tool"
                params = getattr(tool, "input_parameters", None)
                if not params:
                    # Minimal shape if schema missing
                    params = {"type": "object", "properties": {}}
                formatted.append(
                    {
                        "type": "function",
                        "function": {
                            "name": name,
                            "description": description,
                            "parameters": params,
                        },
                    }
                )
            except Exception as e:
                logger.warning("Failed to map Composio tool schema: %s", e)
                continue
        return formatted

    def execute(
        self,
        *,
        slug: str,
        arguments: dict[str, t.Any],
        user_id: str | None = None,
        connected_account_id: str | None = None,
        custom_auth_params: dict[str, t.Any] | None = None,
        custom_connection_data: dict[str, t.Any] | None = None,
        text: str | None = None,
        version: str | None = None,
        toolkit_versions: t.Any | None = None,
        modifiers: t.Any | None = None,
    ) -> dict[str, t.Any]:
        """
        Execute a Composio tool and return a normalized response dict.

        Args:
            slug (str): Tool slug to execute.
            arguments (dict[str, Any]): Arguments for the tool.
            user_id (str | None): Optional user ID.
            connected_account_id (str | None): Optional connected account ID.
            custom_auth_params (dict[str, Any] | None): Optional custom auth params.
            custom_connection_data (dict[str, Any] | None): Optional custom connection data.
            text (str | None): Optional text input.
            version (str | None): Optional version.
            toolkit_versions (Any | None): Optional toolkit versions.
            modifiers (Any | None): Optional modifiers.

        Returns:
            dict[str, Any]: Normalized response dict with keys: successful, data, error.
        """
        resp = self._composio.tools.execute(
            slug=slug,
            arguments=arguments,
            user_id=user_id,
            connected_account_id=connected_account_id,
            custom_auth_params=custom_auth_params,
            custom_connection_data=custom_connection_data,
            text=text,
            version=version,
            toolkit_versions=toolkit_versions,
            modifiers=modifiers,
        )

        # The SDK returns a TypedDict-like object; ensure plain dict
        if hasattr(resp, "copy") and not isinstance(resp, dict):  # e.g., TypedDict proxy
            try:
                resp = dict(resp)  # type: ignore[assignment]
            except Exception as exc:
                logger.debug("Could not coerce Composio response to dict: %s", exc)

        # Normalize key presence
        successful = bool(resp.get("successful", False))  # type: ignore[arg-type]
        data = resp.get("data")
        error = resp.get("error")
        return {"successful": successful, "data": data, "error": error}

Functions

__init__

__init__(*, api_key=None, provider=None, file_download_dir=None, toolkit_versions=None)

Initialize the ComposioAdapter.

Parameters:

Name	Type	Description	Default
api_key	str | None	Optional API key for Composio.	None
provider	Any | None	Optional provider integration.	None
file_download_dir	str | None	Directory for auto file handling.	None
toolkit_versions	Any | None	Toolkit version overrides.	None

Raises:

Type	Description
ImportError	If composio SDK is not installed.

Source code in pyagenity/adapters/tools/composio_adapter.py

def __init__(
    self,
    *,
    api_key: str | None = None,
    provider: t.Any | None = None,
    file_download_dir: str | None = None,
    toolkit_versions: t.Any | None = None,
) -> None:
    """
    Initialize the ComposioAdapter.

    Args:
        api_key (str | None): Optional API key for Composio.
        provider (Any | None): Optional provider integration.
        file_download_dir (str | None): Directory for auto file handling.
        toolkit_versions (Any | None): Toolkit version overrides.

    Raises:
        ImportError: If composio SDK is not installed.
    """
    if not HAS_COMPOSIO:
        raise ImportError(
            "ComposioAdapter requires 'composio' package. Install with: "
            "pip install pyagenity[composio]"
        )

    self._composio = Composio(  # type: ignore[call-arg]
        api_key=api_key,
        provider=provider,
        file_download_dir=file_download_dir,
        toolkit_versions=toolkit_versions,
    )

execute

execute(*, slug, arguments, user_id=None, connected_account_id=None, custom_auth_params=None, custom_connection_data=None, text=None, version=None, toolkit_versions=None, modifiers=None)

Execute a Composio tool and return a normalized response dict.

Parameters:

Name	Type	Description	Default
slug	str	Tool slug to execute.	required
arguments	dict[str, Any]	Arguments for the tool.	required
user_id	str | None	Optional user ID.	None
connected_account_id	str | None	Optional connected account ID.	None
custom_auth_params	dict[str, Any] | None	Optional custom auth params.	None
custom_connection_data	dict[str, Any] | None	Optional custom connection data.	None
text	str | None	Optional text input.	None
version	str | None	Optional version.	None
toolkit_versions	Any | None	Optional toolkit versions.	None
modifiers	Any | None	Optional modifiers.	None

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Normalized response dict with keys: successful, data, error.

Source code in pyagenity/adapters/tools/composio_adapter.py

def execute(
    self,
    *,
    slug: str,
    arguments: dict[str, t.Any],
    user_id: str | None = None,
    connected_account_id: str | None = None,
    custom_auth_params: dict[str, t.Any] | None = None,
    custom_connection_data: dict[str, t.Any] | None = None,
    text: str | None = None,
    version: str | None = None,
    toolkit_versions: t.Any | None = None,
    modifiers: t.Any | None = None,
) -> dict[str, t.Any]:
    """
    Execute a Composio tool and return a normalized response dict.

    Args:
        slug (str): Tool slug to execute.
        arguments (dict[str, Any]): Arguments for the tool.
        user_id (str | None): Optional user ID.
        connected_account_id (str | None): Optional connected account ID.
        custom_auth_params (dict[str, Any] | None): Optional custom auth params.
        custom_connection_data (dict[str, Any] | None): Optional custom connection data.
        text (str | None): Optional text input.
        version (str | None): Optional version.
        toolkit_versions (Any | None): Optional toolkit versions.
        modifiers (Any | None): Optional modifiers.

    Returns:
        dict[str, Any]: Normalized response dict with keys: successful, data, error.
    """
    resp = self._composio.tools.execute(
        slug=slug,
        arguments=arguments,
        user_id=user_id,
        connected_account_id=connected_account_id,
        custom_auth_params=custom_auth_params,
        custom_connection_data=custom_connection_data,
        text=text,
        version=version,
        toolkit_versions=toolkit_versions,
        modifiers=modifiers,
    )

    # The SDK returns a TypedDict-like object; ensure plain dict
    if hasattr(resp, "copy") and not isinstance(resp, dict):  # e.g., TypedDict proxy
        try:
            resp = dict(resp)  # type: ignore[assignment]
        except Exception as exc:
            logger.debug("Could not coerce Composio response to dict: %s", exc)

    # Normalize key presence
    successful = bool(resp.get("successful", False))  # type: ignore[arg-type]
    data = resp.get("data")
    error = resp.get("error")
    return {"successful": successful, "data": data, "error": error}

is_available staticmethod

is_available()

Return True if composio SDK is importable.

Returns:

Name	Type	Description
bool	bool	True if composio SDK is available, False otherwise.

Source code in pyagenity/adapters/tools/composio_adapter.py

@staticmethod
def is_available() -> bool:
    """
    Return True if composio SDK is importable.

    Returns:
        bool: True if composio SDK is available, False otherwise.
    """
    return HAS_COMPOSIO

list_raw_tools_for_llm

list_raw_tools_for_llm(*, tool_slugs=None, toolkits=None, search=None, scopes=None, limit=None)

Return raw Composio tool schemas mapped to function-calling format.

Parameters:

Name	Type	Description	Default
tool_slugs	list[str] | None	Optional list of tool slugs.	None
toolkits	list[str] | None	Optional list of toolkits.	None
search	str | None	Optional search string.	None
scopes	list[str] | None	Optional scopes.	None
limit	int | None	Optional limit on number of tools.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of raw tool schemas in function-calling format.

Source code in pyagenity/adapters/tools/composio_adapter.py

def list_raw_tools_for_llm(
    self,
    *,
    tool_slugs: list[str] | None = None,
    toolkits: list[str] | None = None,
    search: str | None = None,
    scopes: list[str] | None = None,
    limit: int | None = None,
) -> list[dict[str, t.Any]]:
    """
    Return raw Composio tool schemas mapped to function-calling format.

    Args:
        tool_slugs (list[str] | None): Optional list of tool slugs.
        toolkits (list[str] | None): Optional list of toolkits.
        search (str | None): Optional search string.
        scopes (list[str] | None): Optional scopes.
        limit (int | None): Optional limit on number of tools.

    Returns:
        list[dict[str, Any]]: List of raw tool schemas in function-calling format.
    """
    formatted: list[dict[str, t.Any]] = []
    raw_tools = self._composio.tools.get_raw_composio_tools(
        tools=tool_slugs, search=search, toolkits=toolkits, scopes=scopes, limit=limit
    )

    for tool in raw_tools:
        try:
            name = tool.slug  # type: ignore[attr-defined]
            description = getattr(tool, "description", "") or "Composio tool"
            params = getattr(tool, "input_parameters", None)
            if not params:
                # Minimal shape if schema missing
                params = {"type": "object", "properties": {}}
            formatted.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": description,
                        "parameters": params,
                    },
                }
            )
        except Exception as e:
            logger.warning("Failed to map Composio tool schema: %s", e)
            continue
    return formatted

list_tools_for_llm

list_tools_for_llm(*, user_id, tool_slugs=None, toolkits=None, search=None, scopes=None, limit=None)

Return tools formatted for LLM function-calling.

Parameters:

Name	Type	Description	Default
user_id	str	User ID for tool discovery.	required
tool_slugs	list[str] | None	Optional list of tool slugs.	None
toolkits	list[str] | None	Optional list of toolkits.	None
search	str | None	Optional search string.	None
scopes	list[str] | None	Optional scopes.	None
limit	int | None	Optional limit on number of tools.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of tools in function-calling format.

Source code in pyagenity/adapters/tools/composio_adapter.py

def list_tools_for_llm(
    self,
    *,
    user_id: str,
    tool_slugs: list[str] | None = None,
    toolkits: list[str] | None = None,
    search: str | None = None,
    scopes: list[str] | None = None,
    limit: int | None = None,
) -> list[dict[str, t.Any]]:
    """
    Return tools formatted for LLM function-calling.

    Args:
        user_id (str): User ID for tool discovery.
        tool_slugs (list[str] | None): Optional list of tool slugs.
        toolkits (list[str] | None): Optional list of toolkits.
        search (str | None): Optional search string.
        scopes (list[str] | None): Optional scopes.
        limit (int | None): Optional limit on number of tools.

    Returns:
        list[dict[str, Any]]: List of tools in function-calling format.
    """
    # Prefer the provider-wrapped format when available
    tools = self._composio.tools.get(
        user_id=user_id,
        tools=tool_slugs,  # type: ignore[arg-type]
        toolkits=toolkits,  # type: ignore[arg-type]
        search=search,
        scopes=scopes,
        limit=limit,
    )

    # The provider-wrapped output may already be in the desired structure.
    # We'll detect and pass-through; otherwise convert using raw schemas.
    formatted: list[dict[str, t.Any]] = []
    for t_obj in tools if isinstance(tools, list) else []:
        try:
            if (
                isinstance(t_obj, dict)
                and t_obj.get("type") == "function"
                and "function" in t_obj
            ):
                formatted.append(t_obj)
            else:
                # Fallback: try to pull minimal fields
                fn = t_obj.get("function", {}) if isinstance(t_obj, dict) else {}
                if fn.get("name") and fn.get("parameters"):
                    formatted.append({"type": "function", "function": fn})
        except Exception as exc:
            logger.debug("Skipping non-conforming Composio tool wrapper: %s", exc)
            continue

    if formatted:
        return formatted

    # Fallback to raw schemas and convert manually
    formatted.extend(
        self.list_raw_tools_for_llm(
            tool_slugs=tool_slugs, toolkits=toolkits, search=search, scopes=scopes, limit=limit
        )
    )

    return formatted

langchain_adapter

LangChain adapter for PyAgenity (generic wrapper, registry-based).

This adapter mirrors the spirit of Google's ADK LangChain wrapper by allowing you to register any LangChain tool (BaseTool/StructuredTool) or a duck-typed object that exposes a run/_run method, then exposing it to PyAgenity in the uniform function-calling schema that ToolNode expects.

Key points: - Register arbitrary tools at runtime via register_tool / register_tools. - Tool schemas are derived from tool.args (when available) or inferred from the tool's pydantic args_schema; otherwise, we fallback to a minimal best-effort schema inferred from the wrapped function signature. - Execution prefers invoke (Runnable interface) and falls back to run/ _run or calling a wrapped function with kwargs.

Optional install

pip install pyagenity[langchain]

Backward-compat convenience: - For continuity with prior versions, the adapter can auto-register two common tools (tavily_search and requests_get) if autoload_default_tools is True and no user-registered tools exist. You can disable this by passing autoload_default_tools=False to the constructor.

Classes:

Name	Description
LangChainAdapter	Generic registry-based LangChain adapter.
LangChainToolWrapper	Wrap a LangChain tool or a duck-typed tool into a uniform interface.

Attributes:

Name	Type	Description
HAS_LANGCHAIN
logger

Attributes

HAS_LANGCHAIN module-attribute

HAS_LANGCHAIN = find_spec('langchain_core') is not None

logger module-attribute

logger = getLogger(__name__)

Classes

LangChainAdapter

Generic registry-based LangChain adapter.

Notes

• Avoids importing heavy integrations until needed (lazy default autoload).
• Normalizes schemas and execution results into simple dicts.
• Allows arbitrary tool registration instead of hardcoding a tiny set.

Methods:

Name	Description
__init__	Initialize LangChainAdapter.
execute	Execute a supported LangChain tool and normalize the response.
is_available	Return True if langchain-core is importable.
list_tools_for_llm	Return a list of function-calling formatted tool schemas.
register_tool	Register a tool instance and return the resolved name used for exposure.
register_tools	Register multiple tool instances.

Source code in pyagenity/adapters/tools/langchain_adapter.py

class LangChainAdapter:
    """
    Generic registry-based LangChain adapter.

    Notes:
        - Avoids importing heavy integrations until needed (lazy default autoload).
        - Normalizes schemas and execution results into simple dicts.
        - Allows arbitrary tool registration instead of hardcoding a tiny set.
    """

    def __init__(self, *, autoload_default_tools: bool = True) -> None:
        """
        Initialize LangChainAdapter.

        Args:
            autoload_default_tools (bool): Whether to autoload default tools if registry is empty.

        Raises:
            ImportError: If langchain-core is not installed.
        """
        if not HAS_LANGCHAIN:
            raise ImportError(
                "LangChainAdapter requires 'langchain-core' and optional integrations.\n"
                "Install with: pip install pyagenity[langchain]"
            )
        self._registry: dict[str, LangChainToolWrapper] = {}
        self._autoload = autoload_default_tools

    @staticmethod
    def is_available() -> bool:
        """
        Return True if langchain-core is importable.

        Returns:
            bool: True if langchain-core is available, False otherwise.
        """
        return HAS_LANGCHAIN

    # ------------------------
    # Discovery
    # ------------------------
    def list_tools_for_llm(self) -> list[dict[str, t.Any]]:
        """
        Return a list of function-calling formatted tool schemas.

        If registry is empty and autoload is enabled, attempt to autoload a
        couple of common tools for convenience (tavily_search, requests_get).

        Returns:
            list[dict[str, Any]]: List of tool schemas in function-calling format.
        """
        if not self._registry and self._autoload:
            self._try_autoload_defaults()

        return [wrapper.to_schema() for wrapper in self._registry.values()]

    # ------------------------
    # Execute
    # ------------------------
    def execute(self, *, name: str, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
        """
        Execute a supported LangChain tool and normalize the response.

        Args:
            name (str): Name of the tool to execute.
            arguments (dict[str, Any]): Arguments for the tool.

        Returns:
            dict[str, Any]: Normalized response dict with keys: successful, data, error.
        """
        if name not in self._registry and self._autoload:
            # Late autoload attempt in case discovery wasn't called first
            self._try_autoload_defaults()

        wrapper = self._registry.get(name)
        if not wrapper:
            return {"successful": False, "data": None, "error": f"Unknown LangChain tool: {name}"}
        return wrapper.execute(arguments)

    # ------------------------
    # Internals
    # ------------------------
    def register_tool(
        self,
        tool: t.Any,
        *,
        name: str | None = None,
        description: str | None = None,
    ) -> str:
        """
        Register a tool instance and return the resolved name used for exposure.

        Args:
            tool (Any): Tool instance to register.
            name (str | None): Optional override for tool name.
            description (str | None): Optional override for tool description.

        Returns:
            str: The resolved name used for exposure.
        """
        wrapper = LangChainToolWrapper(tool, name=name, description=description)
        self._registry[wrapper.name] = wrapper
        return wrapper.name

    def register_tools(self, tools: list[t.Any]) -> list[str]:
        """
        Register multiple tool instances.

        Args:
            tools (list[Any]): List of tool instances to register.

        Returns:
            list[str]: List of resolved names for the registered tools.
        """
        names: list[str] = []
        for tool in tools:
            names.append(self.register_tool(tool))
        return names

    def _create_tavily_search_tool(self) -> t.Any:
        """
        Construct Tavily search tool lazily.

        Prefer the new dedicated integration `langchain_tavily.TavilySearch`.
        Fall back to the deprecated community tool if needed.

        Returns:
            Any: Tavily search tool instance.

        Raises:
            ImportError: If Tavily tool cannot be imported.
        """
        # Preferred: langchain-tavily
        try:
            mod = importlib.import_module("langchain_tavily")
            return mod.TavilySearch()  # type: ignore[attr-defined]
        except Exception as exc:
            logger.debug("Preferred langchain_tavily import failed: %s", exc)

        # Fallback: deprecated community tool (still functional for now)
        try:
            mod = importlib.import_module("langchain_community.tools.tavily_search")
            return mod.TavilySearchResults()
        except Exception as exc:  # ImportError or runtime
            raise ImportError(
                "Tavily tool requires 'langchain-tavily' (preferred) or"
                " 'langchain-community' with 'tavily-python'.\n"
                "Install with: pip install pyagenity[langchain]"
            ) from exc

    def _create_requests_get_tool(self) -> t.Any:
        """
        Construct RequestsGetTool lazily with a basic requests wrapper.

        Note: Requests tools require an explicit wrapper instance and, for safety,
        default to disallowing dangerous requests. Here we opt-in to allow GET
        requests by setting allow_dangerous_requests=True to make the tool usable
        in agent contexts. Consider tightening this in your application.

        Returns:
            Any: RequestsGetTool instance.

        Raises:
            ImportError: If RequestsGetTool cannot be imported.
        """
        try:
            req_tool_mod = importlib.import_module("langchain_community.tools.requests.tool")
            util_mod = importlib.import_module("langchain_community.utilities.requests")
            wrapper = util_mod.TextRequestsWrapper(headers={})  # type: ignore[attr-defined]
            return req_tool_mod.RequestsGetTool(
                requests_wrapper=wrapper,
                allow_dangerous_requests=True,
            )
        except Exception as exc:  # ImportError or runtime
            raise ImportError(
                "Requests tool requires 'langchain-community'.\n"
                "Install with: pip install pyagenity[langchain]"
            ) from exc

    def _try_autoload_defaults(self) -> None:
        """
        Best-effort autoload of a couple of common tools.

        This keeps prior behavior available while allowing users to register
        arbitrary tools. Failures are logged but non-fatal.

        Returns:
            None
        """
        # Tavily search
        try:
            tavily = self._create_tavily_search_tool()
            self.register_tool(tavily, name="tavily_search")
        except Exception as exc:
            logger.debug("Skipping Tavily autoload: %s", exc)

        # Requests GET
        try:
            rget = self._create_requests_get_tool()
            self.register_tool(rget, name="requests_get")
        except Exception as exc:
            logger.debug("Skipping requests_get autoload: %s", exc)

Functions

__init__

__init__(*, autoload_default_tools=True)

Initialize LangChainAdapter.

Parameters:

Name	Type	Description	Default
autoload_default_tools	bool	Whether to autoload default tools if registry is empty.	True

Raises:

Type	Description
ImportError	If langchain-core is not installed.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def __init__(self, *, autoload_default_tools: bool = True) -> None:
    """
    Initialize LangChainAdapter.

    Args:
        autoload_default_tools (bool): Whether to autoload default tools if registry is empty.

    Raises:
        ImportError: If langchain-core is not installed.
    """
    if not HAS_LANGCHAIN:
        raise ImportError(
            "LangChainAdapter requires 'langchain-core' and optional integrations.\n"
            "Install with: pip install pyagenity[langchain]"
        )
    self._registry: dict[str, LangChainToolWrapper] = {}
    self._autoload = autoload_default_tools

execute

execute(*, name, arguments)

Execute a supported LangChain tool and normalize the response.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool to execute.	required
arguments	dict[str, Any]	Arguments for the tool.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Normalized response dict with keys: successful, data, error.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def execute(self, *, name: str, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
    """
    Execute a supported LangChain tool and normalize the response.

    Args:
        name (str): Name of the tool to execute.
        arguments (dict[str, Any]): Arguments for the tool.

    Returns:
        dict[str, Any]: Normalized response dict with keys: successful, data, error.
    """
    if name not in self._registry and self._autoload:
        # Late autoload attempt in case discovery wasn't called first
        self._try_autoload_defaults()

    wrapper = self._registry.get(name)
    if not wrapper:
        return {"successful": False, "data": None, "error": f"Unknown LangChain tool: {name}"}
    return wrapper.execute(arguments)

is_available staticmethod

is_available()

Return True if langchain-core is importable.

Returns:

Name	Type	Description
bool	bool	True if langchain-core is available, False otherwise.

Source code in pyagenity/adapters/tools/langchain_adapter.py

@staticmethod
def is_available() -> bool:
    """
    Return True if langchain-core is importable.

    Returns:
        bool: True if langchain-core is available, False otherwise.
    """
    return HAS_LANGCHAIN

list_tools_for_llm

list_tools_for_llm()

Return a list of function-calling formatted tool schemas.

If registry is empty and autoload is enabled, attempt to autoload a couple of common tools for convenience (tavily_search, requests_get).

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of tool schemas in function-calling format.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def list_tools_for_llm(self) -> list[dict[str, t.Any]]:
    """
    Return a list of function-calling formatted tool schemas.

    If registry is empty and autoload is enabled, attempt to autoload a
    couple of common tools for convenience (tavily_search, requests_get).

    Returns:
        list[dict[str, Any]]: List of tool schemas in function-calling format.
    """
    if not self._registry and self._autoload:
        self._try_autoload_defaults()

    return [wrapper.to_schema() for wrapper in self._registry.values()]

register_tool

register_tool(tool, *, name=None, description=None)

Register a tool instance and return the resolved name used for exposure.

Parameters:

Name	Type	Description	Default
tool	Any	Tool instance to register.	required
name	str | None	Optional override for tool name.	None
description	str | None	Optional override for tool description.	None

Returns:

Name	Type	Description
str	str	The resolved name used for exposure.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def register_tool(
    self,
    tool: t.Any,
    *,
    name: str | None = None,
    description: str | None = None,
) -> str:
    """
    Register a tool instance and return the resolved name used for exposure.

    Args:
        tool (Any): Tool instance to register.
        name (str | None): Optional override for tool name.
        description (str | None): Optional override for tool description.

    Returns:
        str: The resolved name used for exposure.
    """
    wrapper = LangChainToolWrapper(tool, name=name, description=description)
    self._registry[wrapper.name] = wrapper
    return wrapper.name

register_tools

register_tools(tools)

Register multiple tool instances.

Parameters:

Name	Type	Description	Default
tools	list[Any]	List of tool instances to register.	required

Returns:

Type	Description
list[str]	list[str]: List of resolved names for the registered tools.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def register_tools(self, tools: list[t.Any]) -> list[str]:
    """
    Register multiple tool instances.

    Args:
        tools (list[Any]): List of tool instances to register.

    Returns:
        list[str]: List of resolved names for the registered tools.
    """
    names: list[str] = []
    for tool in tools:
        names.append(self.register_tool(tool))
    return names

LangChainToolWrapper

Wrap a LangChain tool or a duck-typed tool into a uniform interface.

Responsibilities

• Resolve execution entrypoint (invoke/run/_run/callable func)
• Provide a function-calling schema {name, description, parameters}
• Execute with dict arguments and return a JSON-serializable result

Methods:

Name	Description
__init__	Initialize LangChainToolWrapper.
execute	Execute the wrapped tool with the provided arguments.
to_schema	Return the function-calling schema for the wrapped tool.

Attributes:

Name	Type	Description
description
name

Source code in pyagenity/adapters/tools/langchain_adapter.py

class LangChainToolWrapper:
    """
    Wrap a LangChain tool or a duck-typed tool into a uniform interface.

    Responsibilities:
        - Resolve execution entrypoint (invoke/run/_run/callable func)
        - Provide a function-calling schema {name, description, parameters}
        - Execute with dict arguments and return a JSON-serializable result
    """

    def __init__(
        self,
        tool: t.Any,
        *,
        name: str | None = None,
        description: str | None = None,
    ) -> None:
        """
        Initialize LangChainToolWrapper.

        Args:
            tool (Any): The LangChain tool or duck-typed object to wrap.
            name (str | None): Optional override for tool name.
            description (str | None): Optional override for tool description.
        """
        self._tool = tool
        self.name = name or getattr(tool, "name", None) or self._default_name(tool)
        self.description = (
            description
            or getattr(tool, "description", None)
            or f"LangChain tool wrapper for {type(tool).__name__}"
        )
        self._callable = self._resolve_callable(tool)

    @staticmethod
    def _default_name(tool: t.Any) -> str:
        # Prefer class name in snake_case-ish
        cls = type(tool).__name__
        return cls[0].lower() + "".join((c if c.islower() else f"_{c.lower()}") for c in cls[1:])

    @staticmethod
    def _resolve_callable(tool: t.Any) -> t.Callable[..., t.Any] | None:
        # Try StructuredTool.func or coroutine
        try:
            # Avoid importing StructuredTool; duck-type attributes
            if getattr(tool, "func", None) is not None:
                return t.cast(t.Callable[..., t.Any], tool.func)
            if getattr(tool, "coroutine", None) is not None:
                return t.cast(t.Callable[..., t.Any], tool.coroutine)
        except Exception as exc:  # pragma: no cover - defensive
            logger.debug("Ignoring tool callable resolution error: %s", exc)
        # Fallback to run/_run methods as callables
        if hasattr(tool, "_run"):
            return tool._run  # type: ignore[attr-defined]
        if hasattr(tool, "run"):
            return tool.run  # type: ignore[attr-defined]
        # Nothing callable to directly use; rely on invoke/run on execution
        return None

    def _json_schema_from_args_schema(self) -> dict[str, t.Any] | None:
        # LangChain BaseTool typically provides .args (already JSON schema)
        schema = getattr(self._tool, "args", None)
        if isinstance(schema, dict) and schema.get("type") == "object":
            return schema

        # Try args_schema (pydantic v1 or v2)
        args_schema = getattr(self._tool, "args_schema", None)
        if args_schema is None:
            return None
        try:
            # pydantic v2
            if hasattr(args_schema, "model_json_schema"):
                js = args_schema.model_json_schema()  # type: ignore[attr-defined]
            else:  # pydantic v1
                js = args_schema.schema()  # type: ignore[attr-defined]
            # Convert typical pydantic schema to a plain "type: object" with properties
            # Look for properties directly
            props = js.get("properties") or {}
            required = js.get("required") or []
            return {"type": "object", "properties": props, "required": required}
        except Exception:  # pragma: no cover - be tolerant
            return None

    def _infer_schema_from_signature(self) -> dict[str, t.Any]:
        func = self._callable or getattr(self._tool, "invoke", None)
        if func is None or not callable(func):  # last resort empty schema
            return {"type": "object", "properties": {}}

        try:
            sig = inspect.signature(func)
            properties: dict[str, dict[str, t.Any]] = {}
            required: list[str] = []
            for name, param in sig.parameters.items():
                if name in {"self", "run_manager", "config", "callbacks"}:
                    continue
                ann = param.annotation
                json_type: str | None = None
                if ann is not inspect._empty:  # type: ignore[attr-defined]
                    json_type = self._map_annotation_to_json_type(ann)
                prop: dict[str, t.Any] = {}
                if json_type:
                    prop["type"] = json_type
                if param.default is inspect._empty:  # type: ignore[attr-defined]
                    required.append(name)
                properties[name] = prop
            schema: dict[str, t.Any] = {"type": "object", "properties": properties}
            if required:
                schema["required"] = required
            return schema
        except Exception:
            return {"type": "object", "properties": {}}

    @staticmethod
    def _map_annotation_to_json_type(ann: t.Any) -> str | None:
        try:
            origin = t.get_origin(ann) or ann
            mapping = {
                str: "string",
                int: "integer",
                float: "number",
                bool: "boolean",
                list: "array",
                tuple: "array",
                set: "array",
                dict: "object",
            }
            # Typed containers map to base Python containers in get_origin
            return mapping.get(origin)
        except Exception:
            return None

    def to_schema(self) -> dict[str, t.Any]:
        """
        Return the function-calling schema for the wrapped tool.

        Returns:
            dict[str, Any]: Function-calling schema with name, description, parameters.
        """
        schema = self._json_schema_from_args_schema() or self._infer_schema_from_signature()
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": schema,
            },
        }

    def execute(self, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
        """
        Execute the wrapped tool with the provided arguments.

        Args:
            arguments (dict[str, Any]): Arguments to pass to the tool.

        Returns:
            dict[str, Any]: Normalized response dict with keys: successful, data, error.
        """
        try:
            tool = self._tool
            if hasattr(tool, "invoke"):
                result = tool.invoke(arguments)  # type: ignore[misc]
            elif hasattr(tool, "run"):
                result = tool.run(arguments)  # type: ignore[misc]
            elif hasattr(tool, "_run"):
                result = tool._run(arguments)  # type: ignore[attr-defined]
            elif callable(self._callable):
                result = self._callable(**arguments)  # type: ignore[call-arg]
            else:
                raise AttributeError("Tool does not support invoke/run/_run/callable")

            data: t.Any = result
            if not isinstance(result, str | int | float | bool | type(None) | dict | list):
                try:
                    json.dumps(result)
                except Exception:
                    data = str(result)
            return {"successful": True, "data": data, "error": None}
        except Exception as exc:
            logger.error("LangChain wrapped tool '%s' failed: %s", self.name, exc)
            return {"successful": False, "data": None, "error": str(exc)}

Attributes

description instance-attribute

description = description or getattr(tool, 'description', None) or f'LangChain tool wrapper for {__name__}'

name instance-attribute

name = name or getattr(tool, 'name', None) or _default_name(tool)

Functions

__init__

__init__(tool, *, name=None, description=None)

Initialize LangChainToolWrapper.

Parameters:

Name	Type	Description	Default
tool	Any	The LangChain tool or duck-typed object to wrap.	required
name	str | None	Optional override for tool name.	None
description	str | None	Optional override for tool description.	None

Source code in pyagenity/adapters/tools/langchain_adapter.py

def __init__(
    self,
    tool: t.Any,
    *,
    name: str | None = None,
    description: str | None = None,
) -> None:
    """
    Initialize LangChainToolWrapper.

    Args:
        tool (Any): The LangChain tool or duck-typed object to wrap.
        name (str | None): Optional override for tool name.
        description (str | None): Optional override for tool description.
    """
    self._tool = tool
    self.name = name or getattr(tool, "name", None) or self._default_name(tool)
    self.description = (
        description
        or getattr(tool, "description", None)
        or f"LangChain tool wrapper for {type(tool).__name__}"
    )
    self._callable = self._resolve_callable(tool)

execute

execute(arguments)

Execute the wrapped tool with the provided arguments.

Parameters:

Name	Type	Description	Default
arguments	dict[str, Any]	Arguments to pass to the tool.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Normalized response dict with keys: successful, data, error.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def execute(self, arguments: dict[str, t.Any]) -> dict[str, t.Any]:
    """
    Execute the wrapped tool with the provided arguments.

    Args:
        arguments (dict[str, Any]): Arguments to pass to the tool.

    Returns:
        dict[str, Any]: Normalized response dict with keys: successful, data, error.
    """
    try:
        tool = self._tool
        if hasattr(tool, "invoke"):
            result = tool.invoke(arguments)  # type: ignore[misc]
        elif hasattr(tool, "run"):
            result = tool.run(arguments)  # type: ignore[misc]
        elif hasattr(tool, "_run"):
            result = tool._run(arguments)  # type: ignore[attr-defined]
        elif callable(self._callable):
            result = self._callable(**arguments)  # type: ignore[call-arg]
        else:
            raise AttributeError("Tool does not support invoke/run/_run/callable")

        data: t.Any = result
        if not isinstance(result, str | int | float | bool | type(None) | dict | list):
            try:
                json.dumps(result)
            except Exception:
                data = str(result)
        return {"successful": True, "data": data, "error": None}
    except Exception as exc:
        logger.error("LangChain wrapped tool '%s' failed: %s", self.name, exc)
        return {"successful": False, "data": None, "error": str(exc)}

to_schema

to_schema()

Return the function-calling schema for the wrapped tool.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Function-calling schema with name, description, parameters.

Source code in pyagenity/adapters/tools/langchain_adapter.py

def to_schema(self) -> dict[str, t.Any]:
    """
    Return the function-calling schema for the wrapped tool.

    Returns:
        dict[str, Any]: Function-calling schema with name, description, parameters.
    """
    schema = self._json_schema_from_args_schema() or self._infer_schema_from_signature()
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": schema,
        },
    }