Integration adapters for optional third-party SDKs.

This package provides unified wrappers and converters for integrating external tool registries, LLM SDKs, and other third-party services with PyAgenity agent graphs. Adapters expose registry-based discovery, function-calling schemas, and normalized execution for supported providers.

Modules:

Name	Description
llm	Integration adapters for optional third-party LLM SDKs.
tools	Integration adapters for optional third-party SDKs.

Modules

llm

Integration adapters for optional third-party LLM SDKs.

This module exposes universal converter APIs to normalize responses and streaming outputs from popular LLM SDKs (e.g., LiteLLM, OpenAI) for use in PyAgenity agent graphs. Adapters provide unified conversion, streaming, and message normalization for agent workflows.

Exports

BaseConverter: Abstract base class for LLM response converters. ConverterType: Enum of supported converter types. LiteLLMConverter: Converter for LiteLLM responses and streams.

OpenAIConverter: (commented, available if implemented)

Modules:

Name	Description
base_converter
litellm_converter
model_response_converter

Classes:

Name	Description
BaseConverter	Abstract base class for all LLM response converters.
ConverterType	Enumeration of supported converter types for LLM responses.
LiteLLMConverter	Converter for LiteLLM responses to PyAgenity Message format.

Attributes

__all__ module-attribute

__all__ = ['BaseConverter', 'ConverterType', 'LiteLLMConverter']

Classes

BaseConverter

Bases: ABC

Abstract base class for all LLM response converters.

Subclasses should implement methods to convert standard and streaming LLM responses into PyAgenity's internal message/event formats.

Attributes:

Name	Type	Description
state	AgentState | None	Optional agent state for context during conversion.

Methods:

Name	Description
__init__	Initialize the converter.
convert_response	Convert a standard agent response to a Message.
convert_streaming_response	Convert a streaming agent response to an async generator of EventModel or Message.

Source code in pyagenity/adapters/llm/base_converter.py

class BaseConverter(ABC):
    """
    Abstract base class for all LLM response converters.

    Subclasses should implement methods to convert standard and streaming
    LLM responses into PyAgenity's internal message/event formats.

    Attributes:
        state (AgentState | None): Optional agent state for context during conversion.
    """

    def __init__(self, state: AgentState | None = None) -> None:
        """
        Initialize the converter.

        Args:
            state (AgentState | None): Optional agent state for context during conversion.
        """
        self.state = state

    @abstractmethod
    async def convert_response(self, response: Any) -> Message:
        """
        Convert a standard agent response to a Message.

        Args:
            response (Any): The raw response from the LLM or agent.

        Returns:
            Message: The converted message object.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Conversion not implemented for this converter")

    @abstractmethod
    async def convert_streaming_response(
        self,
        config: dict,
        node_name: str,
        response: Any,
        meta: dict | None = None,
    ) -> AsyncGenerator[EventModel | Message, None]:
        """
        Convert a streaming agent response to an async generator of EventModel or Message.

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            response (Any): The raw streaming response from the LLM or agent.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            EventModel | Message: Chunks of the converted streaming response.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Streaming not implemented for this converter")

Attributes

state instance-attribute

state = state

Functions

__init__

__init__(state=None)

Initialize the converter.

Parameters:

Name	Type	Description	Default
state	AgentState | None	Optional agent state for context during conversion.	None

Source code in pyagenity/adapters/llm/base_converter.py

def __init__(self, state: AgentState | None = None) -> None:
    """
    Initialize the converter.

    Args:
        state (AgentState | None): Optional agent state for context during conversion.
    """
    self.state = state

convert_response abstractmethod async

convert_response(response)

Convert a standard agent response to a Message.

Parameters:

Name	Type	Description	Default
response	Any	The raw response from the LLM or agent.	required

Returns:

Name	Type	Description
Message	Message	The converted message object.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_response(self, response: Any) -> Message:
    """
    Convert a standard agent response to a Message.

    Args:
        response (Any): The raw response from the LLM or agent.

    Returns:
        Message: The converted message object.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Conversion not implemented for this converter")

convert_streaming_response abstractmethod async

convert_streaming_response(config, node_name, response, meta=None)

Convert a streaming agent response to an async generator of EventModel or Message.

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters.	required
node_name	str	Name of the node processing the response.	required
response	Any	The raw streaming response from the LLM or agent.	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Type	Description
AsyncGenerator[EventModel | Message, None]	EventModel | Message: Chunks of the converted streaming response.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_streaming_response(
    self,
    config: dict,
    node_name: str,
    response: Any,
    meta: dict | None = None,
) -> AsyncGenerator[EventModel | Message, None]:
    """
    Convert a streaming agent response to an async generator of EventModel or Message.

    Args:
        config (dict): Node configuration parameters.
        node_name (str): Name of the node processing the response.
        response (Any): The raw streaming response from the LLM or agent.
        meta (dict | None): Optional metadata for conversion.

    Yields:
        EventModel | Message: Chunks of the converted streaming response.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Streaming not implemented for this converter")

ConverterType

Bases: Enum

Enumeration of supported converter types for LLM responses.

Attributes:

Name	Type	Description
ANTHROPIC
CUSTOM
GOOGLE
LITELLM
OPENAI

Source code in pyagenity/adapters/llm/base_converter.py

class ConverterType(Enum):
    """Enumeration of supported converter types for LLM responses."""

    OPENAI = "openai"
    LITELLM = "litellm"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    CUSTOM = "custom"

Attributes

ANTHROPIC class-attribute instance-attribute

ANTHROPIC = 'anthropic'

CUSTOM class-attribute instance-attribute

CUSTOM = 'custom'

GOOGLE class-attribute instance-attribute

GOOGLE = 'google'

LITELLM class-attribute instance-attribute

LITELLM = 'litellm'

OPENAI class-attribute instance-attribute

OPENAI = 'openai'

LiteLLMConverter

Bases: BaseConverter

Converter for LiteLLM responses to PyAgenity Message format.

Handles both standard and streaming responses, extracting content, reasoning, tool calls, and token usage details.

Methods:

Name	Description
__init__	Initialize the converter.
convert_response	Convert a LiteLLM ModelResponse to a Message.
convert_streaming_response	Convert a LiteLLM streaming or standard response to Message(s).

Attributes:

Name	Type	Description
state

Source code in pyagenity/adapters/llm/litellm_converter.py

class LiteLLMConverter(BaseConverter):
    """
    Converter for LiteLLM responses to PyAgenity Message format.

    Handles both standard and streaming responses, extracting content, reasoning,
    tool calls, and token usage details.
    """

    async def convert_response(self, response: ModelResponse) -> Message:
        """
        Convert a LiteLLM ModelResponse to a Message.

        Args:
            response (ModelResponse): The LiteLLM model response object.

        Returns:
            Message: The converted message object.

        Raises:
            ImportError: If LiteLLM is not installed.
        """
        if not HAS_LITELLM:
            raise ImportError("litellm is not installed. Please install it to use this converter.")

        data = response.model_dump()

        usages_data = data.get("usage", {})

        usages = TokenUsages(
            completion_tokens=usages_data.get("completion_tokens", 0),
            prompt_tokens=usages_data.get("prompt_tokens", 0),
            total_tokens=usages_data.get("total_tokens", 0),
            cache_creation_input_tokens=usages_data.get("cache_creation_input_tokens", 0),
            cache_read_input_tokens=usages_data.get("cache_read_input_tokens", 0),
            reasoning_tokens=usages_data.get("prompt_tokens_details", {}).get(
                "reasoning_tokens", 0
            ),
        )

        created_date = data.get("created", datetime.now())

        # Extract tool calls from response
        tools_calls = data.get("choices", [{}])[0].get("message", {}).get("tool_calls", []) or []

        logger.debug("Creating message from model response with id: %s", response.id)
        content = data.get("choices", [{}])[0].get("message", {}).get("content", "") or ""
        reasoning_content = (
            data.get("choices", [{}])[0].get("message", {}).get("reasoning_content", "") or ""
        )

        blocks = []
        if content:
            blocks.append(TextBlock(text=content))
        if reasoning_content:
            blocks.append(ReasoningBlock(summary=reasoning_content))
        final_tool_calls = []
        for tool_call in tools_calls:
            tool_id = tool_call.get("id", None)
            args = tool_call.get("function", {}).get("arguments", None)
            name = tool_call.get("function", {}).get("name", None)

            if not tool_id or not args or not name:
                continue

            blocks.append(
                ToolCallBlock(
                    name=name,
                    args=json.loads(args),
                    id=tool_id,
                )
            )

            if hasattr(tool_call, "model_dump"):
                final_tool_calls.append(tool_call.model_dump())
            else:
                final_tool_calls.append(tool_call)

        return Message(
            message_id=generate_id(response.id),
            role="assistant",
            content=blocks,
            reasoning=reasoning_content,
            timestamp=created_date,
            metadata={
                "provider": "litellm",
                "model": data.get("model", ""),
                "finish_reason": data.get("choices", [{}])[0].get("finish_reason", "UNKNOWN"),
                "object": data.get("object", ""),
                "prompt_tokens_details": usages_data.get("prompt_tokens_details", {}),
                "completion_tokens_details": usages_data.get("completion_tokens_details", {}),
            },
            usages=usages,
            raw=data,
            tools_calls=final_tool_calls if final_tool_calls else None,
        )

    def _process_chunk(
        self,
        chunk: ModelResponseStream | None,
        seq: int,
        accumulated_content: str,
        accumulated_reasoning_content: str,
        tool_calls: list,
        tool_ids: set,
    ) -> tuple[str, str, list, int, Message | None]:
        """
        Process a single chunk from a LiteLLM streaming response.

        Args:
            chunk (ModelResponseStream | None): The current chunk from the stream.
            seq (int): Sequence number of the chunk.
            accumulated_content (str): Accumulated text content so far.
            accumulated_reasoning_content (str): Accumulated reasoning content so far.
            tool_calls (list): List of tool calls detected so far.
            tool_ids (set): Set of tool call IDs to avoid duplicates.

        Returns:
            tuple: Updated accumulated content, reasoning, tool calls, sequence,
                and Message (if any).
        """
        if not chunk:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None

        msg: ModelResponseStream = chunk  # type: ignore
        if msg is None:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None
        if msg.choices is None or len(msg.choices) == 0:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None
        delta = msg.choices[0].delta
        if delta is None:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None

        # update text delta
        text_part = delta.content or ""
        content_blocks = []
        if text_part:
            content_blocks.append(TextBlock(text=text_part))
        reasoning_part = getattr(delta, "reasoning_content", "") or ""
        if reasoning_part:
            content_blocks.append(ReasoningBlock(summary=reasoning_part))
        accumulated_content += text_part
        accumulated_reasoning_content += reasoning_part
        # handle tool calls if present
        if getattr(delta, "tool_calls", None):
            for tc in delta.tool_calls:  # type: ignore[attr-defined]
                if not tc:
                    continue
                if tc.id in tool_ids:
                    continue
                tool_ids.add(tc.id)
                tool_calls.append(tc.model_dump())
                content_blocks.append(
                    ToolCallBlock(
                        name=tc.function.name,  # type: ignore
                        args=json.loads(tc.function.arguments),  # type: ignore
                        id=tc.id,  # type: ignore
                    )
                )

        output_message = Message(
            message_id=generate_id(msg.id),
            role="assistant",
            content=content_blocks,
            reasoning=accumulated_reasoning_content,
            tools_calls=tool_calls,
            delta=True,
        )

        return accumulated_content, accumulated_reasoning_content, tool_calls, seq, output_message

    async def _handle_stream(
        self,
        config: dict,
        node_name: str,
        stream: CustomStreamWrapper,
        meta: dict | None = None,
    ) -> AsyncGenerator[Message]:
        """
        Handle a LiteLLM streaming response and yield Message objects for each chunk.

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            stream (CustomStreamWrapper): The LiteLLM streaming response object.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            Message: Converted message chunk from the stream.
        """
        accumulated_content = ""
        tool_calls = []
        tool_ids = set()
        accumulated_reasoning_content = ""
        seq = 0

        is_awaitable = inspect.isawaitable(stream)

        # Await stream if necessary
        if is_awaitable:
            stream = await stream

        # Try async iteration (acompletion)
        try:
            async for chunk in stream:
                accumulated_content, accumulated_reasoning_content, tool_calls, seq, message = (
                    self._process_chunk(
                        chunk,
                        seq,
                        accumulated_content,
                        accumulated_reasoning_content,
                        tool_calls,
                        tool_ids,
                    )
                )

                if message:
                    yield message
        except Exception:  # noqa: S110 # nosec B110
            pass

        # Try sync iteration (completion)
        try:
            for chunk in stream:
                accumulated_content, accumulated_reasoning_content, tool_calls, seq, message = (
                    self._process_chunk(
                        chunk,
                        seq,
                        accumulated_content,
                        accumulated_reasoning_content,
                        tool_calls,
                        tool_ids,
                    )
                )

                if message:
                    yield message
        except Exception:  # noqa: S110 # nosec B110
            pass

        # After streaming, yield final message
        metadata = meta or {}
        metadata["provider"] = "litellm"
        metadata["node_name"] = node_name
        metadata["thread_id"] = config.get("thread_id")

        blocks = []
        if accumulated_content:
            blocks.append(TextBlock(text=accumulated_content))
        if accumulated_reasoning_content:
            blocks.append(ReasoningBlock(summary=accumulated_reasoning_content))
        if tool_calls:
            for tc in tool_calls:
                blocks.append(
                    ToolCallBlock(
                        name=tc.get("function", {}).get("name", ""),
                        args=json.loads(tc.get("function", {}).get("arguments", "{}")),
                        id=tc.get("id", ""),
                    )
                )

        logger.debug(
            "Loop done Content: %s  Reasoning: %s Tool Calls: %s",
            accumulated_content,
            accumulated_reasoning_content,
            len(tool_calls),
        )
        message = Message(
            role="assistant",
            message_id=generate_id(None),
            content=blocks,
            delta=False,
            reasoning=accumulated_reasoning_content,
            tools_calls=tool_calls,
            metadata=metadata,
        )
        yield message

    async def convert_streaming_response(  # type: ignore
        self,
        config: dict,
        node_name: str,
        response: Any,
        meta: dict | None = None,
    ) -> AsyncGenerator[Message]:
        """
        Convert a LiteLLM streaming or standard response to Message(s).

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            response (Any): The LiteLLM response object (stream or standard).
            meta (dict | None): Optional metadata for conversion.

        Yields:
            Message: Converted message(s) from the response.

        Raises:
            ImportError: If LiteLLM is not installed.
            Exception: If response type is unsupported.
        """
        if not HAS_LITELLM:
            raise ImportError("litellm is not installed. Please install it to use this converter.")

        if isinstance(response, CustomStreamWrapper):  # type: ignore[possibly-unbound]
            stream = cast(CustomStreamWrapper, response)
            async for event in self._handle_stream(
                config or {},
                node_name or "",
                stream,
                meta,
            ):
                yield event
        elif isinstance(response, ModelResponse):  # type: ignore[possibly-unbound]
            message = await self.convert_response(cast(ModelResponse, response))
            yield message
        else:
            raise Exception("Unsupported response type for LiteLLMConverter")

Attributes

state instance-attribute

state = state

Functions

__init__

__init__(state=None)

Initialize the converter.

Parameters:

Name	Type	Description	Default
state	AgentState | None	Optional agent state for context during conversion.	None

Source code in pyagenity/adapters/llm/base_converter.py

def __init__(self, state: AgentState | None = None) -> None:
    """
    Initialize the converter.

    Args:
        state (AgentState | None): Optional agent state for context during conversion.
    """
    self.state = state

convert_response async

convert_response(response)

Convert a LiteLLM ModelResponse to a Message.

Parameters:

Name	Type	Description	Default
response	ModelResponse	The LiteLLM model response object.	required

Returns:

Name	Type	Description
Message	Message	The converted message object.

Raises:

Type	Description
ImportError	If LiteLLM is not installed.

Source code in pyagenity/adapters/llm/litellm_converter.py

async def convert_response(self, response: ModelResponse) -> Message:
    """
    Convert a LiteLLM ModelResponse to a Message.

    Args:
        response (ModelResponse): The LiteLLM model response object.

    Returns:
        Message: The converted message object.

    Raises:
        ImportError: If LiteLLM is not installed.
    """
    if not HAS_LITELLM:
        raise ImportError("litellm is not installed. Please install it to use this converter.")

    data = response.model_dump()

    usages_data = data.get("usage", {})

    usages = TokenUsages(
        completion_tokens=usages_data.get("completion_tokens", 0),
        prompt_tokens=usages_data.get("prompt_tokens", 0),
        total_tokens=usages_data.get("total_tokens", 0),
        cache_creation_input_tokens=usages_data.get("cache_creation_input_tokens", 0),
        cache_read_input_tokens=usages_data.get("cache_read_input_tokens", 0),
        reasoning_tokens=usages_data.get("prompt_tokens_details", {}).get(
            "reasoning_tokens", 0
        ),
    )

    created_date = data.get("created", datetime.now())

    # Extract tool calls from response
    tools_calls = data.get("choices", [{}])[0].get("message", {}).get("tool_calls", []) or []

    logger.debug("Creating message from model response with id: %s", response.id)
    content = data.get("choices", [{}])[0].get("message", {}).get("content", "") or ""
    reasoning_content = (
        data.get("choices", [{}])[0].get("message", {}).get("reasoning_content", "") or ""
    )

    blocks = []
    if content:
        blocks.append(TextBlock(text=content))
    if reasoning_content:
        blocks.append(ReasoningBlock(summary=reasoning_content))
    final_tool_calls = []
    for tool_call in tools_calls:
        tool_id = tool_call.get("id", None)
        args = tool_call.get("function", {}).get("arguments", None)
        name = tool_call.get("function", {}).get("name", None)

        if not tool_id or not args or not name:
            continue

        blocks.append(
            ToolCallBlock(
                name=name,
                args=json.loads(args),
                id=tool_id,
            )
        )

        if hasattr(tool_call, "model_dump"):
            final_tool_calls.append(tool_call.model_dump())
        else:
            final_tool_calls.append(tool_call)

    return Message(
        message_id=generate_id(response.id),
        role="assistant",
        content=blocks,
        reasoning=reasoning_content,
        timestamp=created_date,
        metadata={
            "provider": "litellm",
            "model": data.get("model", ""),
            "finish_reason": data.get("choices", [{}])[0].get("finish_reason", "UNKNOWN"),
            "object": data.get("object", ""),
            "prompt_tokens_details": usages_data.get("prompt_tokens_details", {}),
            "completion_tokens_details": usages_data.get("completion_tokens_details", {}),
        },
        usages=usages,
        raw=data,
        tools_calls=final_tool_calls if final_tool_calls else None,
    )

convert_streaming_response async

convert_streaming_response(config, node_name, response, meta=None)

Convert a LiteLLM streaming or standard response to Message(s).

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters.	required
node_name	str	Name of the node processing the response.	required
response	Any	The LiteLLM response object (stream or standard).	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Name	Type	Description
Message	AsyncGenerator[Message]	Converted message(s) from the response.

Raises:

Type	Description
ImportError	If LiteLLM is not installed.
Exception	If response type is unsupported.

Source code in pyagenity/adapters/llm/litellm_converter.py

async def convert_streaming_response(  # type: ignore
    self,
    config: dict,
    node_name: str,
    response: Any,
    meta: dict | None = None,
) -> AsyncGenerator[Message]:
    """
    Convert a LiteLLM streaming or standard response to Message(s).

    Args:
        config (dict): Node configuration parameters.
        node_name (str): Name of the node processing the response.
        response (Any): The LiteLLM response object (stream or standard).
        meta (dict | None): Optional metadata for conversion.

    Yields:
        Message: Converted message(s) from the response.

    Raises:
        ImportError: If LiteLLM is not installed.
        Exception: If response type is unsupported.
    """
    if not HAS_LITELLM:
        raise ImportError("litellm is not installed. Please install it to use this converter.")

    if isinstance(response, CustomStreamWrapper):  # type: ignore[possibly-unbound]
        stream = cast(CustomStreamWrapper, response)
        async for event in self._handle_stream(
            config or {},
            node_name or "",
            stream,
            meta,
        ):
            yield event
    elif isinstance(response, ModelResponse):  # type: ignore[possibly-unbound]
        message = await self.convert_response(cast(ModelResponse, response))
        yield message
    else:
        raise Exception("Unsupported response type for LiteLLMConverter")

Modules

base_converter

Classes:

Name	Description
BaseConverter	Abstract base class for all LLM response converters.
ConverterType	Enumeration of supported converter types for LLM responses.

Classes

BaseConverter

Bases: ABC

Abstract base class for all LLM response converters.

Subclasses should implement methods to convert standard and streaming LLM responses into PyAgenity's internal message/event formats.

Attributes:

Name	Type	Description
state	AgentState | None	Optional agent state for context during conversion.

Methods:

Name	Description
__init__	Initialize the converter.
convert_response	Convert a standard agent response to a Message.
convert_streaming_response	Convert a streaming agent response to an async generator of EventModel or Message.

Source code in pyagenity/adapters/llm/base_converter.py

class BaseConverter(ABC):
    """
    Abstract base class for all LLM response converters.

    Subclasses should implement methods to convert standard and streaming
    LLM responses into PyAgenity's internal message/event formats.

    Attributes:
        state (AgentState | None): Optional agent state for context during conversion.
    """

    def __init__(self, state: AgentState | None = None) -> None:
        """
        Initialize the converter.

        Args:
            state (AgentState | None): Optional agent state for context during conversion.
        """
        self.state = state

    @abstractmethod
    async def convert_response(self, response: Any) -> Message:
        """
        Convert a standard agent response to a Message.

        Args:
            response (Any): The raw response from the LLM or agent.

        Returns:
            Message: The converted message object.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Conversion not implemented for this converter")

    @abstractmethod
    async def convert_streaming_response(
        self,
        config: dict,
        node_name: str,
        response: Any,
        meta: dict | None = None,
    ) -> AsyncGenerator[EventModel | Message, None]:
        """
        Convert a streaming agent response to an async generator of EventModel or Message.

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            response (Any): The raw streaming response from the LLM or agent.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            EventModel | Message: Chunks of the converted streaming response.

        Raises:
            NotImplementedError: If not implemented in subclass.
        """
        raise NotImplementedError("Streaming not implemented for this converter")

Attributes

state instance-attribute

state = state

Functions

__init__

__init__(state=None)

Initialize the converter.

Parameters:

Name	Type	Description	Default
state	AgentState | None	Optional agent state for context during conversion.	None

Source code in pyagenity/adapters/llm/base_converter.py

def __init__(self, state: AgentState | None = None) -> None:
    """
    Initialize the converter.

    Args:
        state (AgentState | None): Optional agent state for context during conversion.
    """
    self.state = state

convert_response abstractmethod async

convert_response(response)

Convert a standard agent response to a Message.

Parameters:

Name	Type	Description	Default
response	Any	The raw response from the LLM or agent.	required

Returns:

Name	Type	Description
Message	Message	The converted message object.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_response(self, response: Any) -> Message:
    """
    Convert a standard agent response to a Message.

    Args:
        response (Any): The raw response from the LLM or agent.

    Returns:
        Message: The converted message object.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Conversion not implemented for this converter")

convert_streaming_response abstractmethod async

convert_streaming_response(config, node_name, response, meta=None)

Convert a streaming agent response to an async generator of EventModel or Message.

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters.	required
node_name	str	Name of the node processing the response.	required
response	Any	The raw streaming response from the LLM or agent.	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Type	Description
AsyncGenerator[EventModel | Message, None]	EventModel | Message: Chunks of the converted streaming response.

Raises:

Type	Description
NotImplementedError	If not implemented in subclass.

Source code in pyagenity/adapters/llm/base_converter.py

@abstractmethod
async def convert_streaming_response(
    self,
    config: dict,
    node_name: str,
    response: Any,
    meta: dict | None = None,
) -> AsyncGenerator[EventModel | Message, None]:
    """
    Convert a streaming agent response to an async generator of EventModel or Message.

    Args:
        config (dict): Node configuration parameters.
        node_name (str): Name of the node processing the response.
        response (Any): The raw streaming response from the LLM or agent.
        meta (dict | None): Optional metadata for conversion.

    Yields:
        EventModel | Message: Chunks of the converted streaming response.

    Raises:
        NotImplementedError: If not implemented in subclass.
    """
    raise NotImplementedError("Streaming not implemented for this converter")

ConverterType

Bases: Enum

Enumeration of supported converter types for LLM responses.

Attributes:

Name	Type	Description
ANTHROPIC
CUSTOM
GOOGLE
LITELLM
OPENAI

Source code in pyagenity/adapters/llm/base_converter.py

class ConverterType(Enum):
    """Enumeration of supported converter types for LLM responses."""

    OPENAI = "openai"
    LITELLM = "litellm"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    CUSTOM = "custom"

Attributes

ANTHROPIC class-attribute instance-attribute

ANTHROPIC = 'anthropic'

CUSTOM class-attribute instance-attribute

CUSTOM = 'custom'

GOOGLE class-attribute instance-attribute

GOOGLE = 'google'

LITELLM class-attribute instance-attribute

LITELLM = 'litellm'

OPENAI class-attribute instance-attribute

OPENAI = 'openai'

litellm_converter

Classes:

Name	Description
LiteLLMConverter	Converter for LiteLLM responses to PyAgenity Message format.

Attributes:

Name	Type	Description
HAS_LITELLM
logger

Attributes

HAS_LITELLM module-attribute

HAS_LITELLM = True

logger module-attribute

logger = getLogger(__name__)

Classes

LiteLLMConverter

Bases: BaseConverter

Converter for LiteLLM responses to PyAgenity Message format.

Handles both standard and streaming responses, extracting content, reasoning, tool calls, and token usage details.

Methods:

Name	Description
__init__	Initialize the converter.
convert_response	Convert a LiteLLM ModelResponse to a Message.
convert_streaming_response	Convert a LiteLLM streaming or standard response to Message(s).

Attributes:

Name	Type	Description
state

Source code in pyagenity/adapters/llm/litellm_converter.py

class LiteLLMConverter(BaseConverter):
    """
    Converter for LiteLLM responses to PyAgenity Message format.

    Handles both standard and streaming responses, extracting content, reasoning,
    tool calls, and token usage details.
    """

    async def convert_response(self, response: ModelResponse) -> Message:
        """
        Convert a LiteLLM ModelResponse to a Message.

        Args:
            response (ModelResponse): The LiteLLM model response object.

        Returns:
            Message: The converted message object.

        Raises:
            ImportError: If LiteLLM is not installed.
        """
        if not HAS_LITELLM:
            raise ImportError("litellm is not installed. Please install it to use this converter.")

        data = response.model_dump()

        usages_data = data.get("usage", {})

        usages = TokenUsages(
            completion_tokens=usages_data.get("completion_tokens", 0),
            prompt_tokens=usages_data.get("prompt_tokens", 0),
            total_tokens=usages_data.get("total_tokens", 0),
            cache_creation_input_tokens=usages_data.get("cache_creation_input_tokens", 0),
            cache_read_input_tokens=usages_data.get("cache_read_input_tokens", 0),
            reasoning_tokens=usages_data.get("prompt_tokens_details", {}).get(
                "reasoning_tokens", 0
            ),
        )

        created_date = data.get("created", datetime.now())

        # Extract tool calls from response
        tools_calls = data.get("choices", [{}])[0].get("message", {}).get("tool_calls", []) or []

        logger.debug("Creating message from model response with id: %s", response.id)
        content = data.get("choices", [{}])[0].get("message", {}).get("content", "") or ""
        reasoning_content = (
            data.get("choices", [{}])[0].get("message", {}).get("reasoning_content", "") or ""
        )

        blocks = []
        if content:
            blocks.append(TextBlock(text=content))
        if reasoning_content:
            blocks.append(ReasoningBlock(summary=reasoning_content))
        final_tool_calls = []
        for tool_call in tools_calls:
            tool_id = tool_call.get("id", None)
            args = tool_call.get("function", {}).get("arguments", None)
            name = tool_call.get("function", {}).get("name", None)

            if not tool_id or not args or not name:
                continue

            blocks.append(
                ToolCallBlock(
                    name=name,
                    args=json.loads(args),
                    id=tool_id,
                )
            )

            if hasattr(tool_call, "model_dump"):
                final_tool_calls.append(tool_call.model_dump())
            else:
                final_tool_calls.append(tool_call)

        return Message(
            message_id=generate_id(response.id),
            role="assistant",
            content=blocks,
            reasoning=reasoning_content,
            timestamp=created_date,
            metadata={
                "provider": "litellm",
                "model": data.get("model", ""),
                "finish_reason": data.get("choices", [{}])[0].get("finish_reason", "UNKNOWN"),
                "object": data.get("object", ""),
                "prompt_tokens_details": usages_data.get("prompt_tokens_details", {}),
                "completion_tokens_details": usages_data.get("completion_tokens_details", {}),
            },
            usages=usages,
            raw=data,
            tools_calls=final_tool_calls if final_tool_calls else None,
        )

    def _process_chunk(
        self,
        chunk: ModelResponseStream | None,
        seq: int,
        accumulated_content: str,
        accumulated_reasoning_content: str,
        tool_calls: list,
        tool_ids: set,
    ) -> tuple[str, str, list, int, Message | None]:
        """
        Process a single chunk from a LiteLLM streaming response.

        Args:
            chunk (ModelResponseStream | None): The current chunk from the stream.
            seq (int): Sequence number of the chunk.
            accumulated_content (str): Accumulated text content so far.
            accumulated_reasoning_content (str): Accumulated reasoning content so far.
            tool_calls (list): List of tool calls detected so far.
            tool_ids (set): Set of tool call IDs to avoid duplicates.

        Returns:
            tuple: Updated accumulated content, reasoning, tool calls, sequence,
                and Message (if any).
        """
        if not chunk:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None

        msg: ModelResponseStream = chunk  # type: ignore
        if msg is None:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None
        if msg.choices is None or len(msg.choices) == 0:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None
        delta = msg.choices[0].delta
        if delta is None:
            return accumulated_content, accumulated_reasoning_content, tool_calls, seq, None

        # update text delta
        text_part = delta.content or ""
        content_blocks = []
        if text_part:
            content_blocks.append(TextBlock(text=text_part))
        reasoning_part = getattr(delta, "reasoning_content", "") or ""
        if reasoning_part:
            content_blocks.append(ReasoningBlock(summary=reasoning_part))
        accumulated_content += text_part
        accumulated_reasoning_content += reasoning_part
        # handle tool calls if present
        if getattr(delta, "tool_calls", None):
            for tc in delta.tool_calls:  # type: ignore[attr-defined]
                if not tc:
                    continue
                if tc.id in tool_ids:
                    continue
                tool_ids.add(tc.id)
                tool_calls.append(tc.model_dump())
                content_blocks.append(
                    ToolCallBlock(
                        name=tc.function.name,  # type: ignore
                        args=json.loads(tc.function.arguments),  # type: ignore
                        id=tc.id,  # type: ignore
                    )
                )

        output_message = Message(
            message_id=generate_id(msg.id),
            role="assistant",
            content=content_blocks,
            reasoning=accumulated_reasoning_content,
            tools_calls=tool_calls,
            delta=True,
        )

        return accumulated_content, accumulated_reasoning_content, tool_calls, seq, output_message

    async def _handle_stream(
        self,
        config: dict,
        node_name: str,
        stream: CustomStreamWrapper,
        meta: dict | None = None,
    ) -> AsyncGenerator[Message]:
        """
        Handle a LiteLLM streaming response and yield Message objects for each chunk.

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            stream (CustomStreamWrapper): The LiteLLM streaming response object.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            Message: Converted message chunk from the stream.
        """
        accumulated_content = ""
        tool_calls = []
        tool_ids = set()
        accumulated_reasoning_content = ""
        seq = 0

        is_awaitable = inspect.isawaitable(stream)

        # Await stream if necessary
        if is_awaitable:
            stream = await stream

        # Try async iteration (acompletion)
        try:
            async for chunk in stream:
                accumulated_content, accumulated_reasoning_content, tool_calls, seq, message = (
                    self._process_chunk(
                        chunk,
                        seq,
                        accumulated_content,
                        accumulated_reasoning_content,
                        tool_calls,
                        tool_ids,
                    )
                )

                if message:
                    yield message
        except Exception:  # noqa: S110 # nosec B110
            pass

        # Try sync iteration (completion)
        try:
            for chunk in stream:
                accumulated_content, accumulated_reasoning_content, tool_calls, seq, message = (
                    self._process_chunk(
                        chunk,
                        seq,
                        accumulated_content,
                        accumulated_reasoning_content,
                        tool_calls,
                        tool_ids,
                    )
                )

                if message:
                    yield message
        except Exception:  # noqa: S110 # nosec B110
            pass

        # After streaming, yield final message
        metadata = meta or {}
        metadata["provider"] = "litellm"
        metadata["node_name"] = node_name
        metadata["thread_id"] = config.get("thread_id")

        blocks = []
        if accumulated_content:
            blocks.append(TextBlock(text=accumulated_content))
        if accumulated_reasoning_content:
            blocks.append(ReasoningBlock(summary=accumulated_reasoning_content))
        if tool_calls:
            for tc in tool_calls:
                blocks.append(
                    ToolCallBlock(
                        name=tc.get("function", {}).get("name", ""),
                        args=json.loads(tc.get("function", {}).get("arguments", "{}")),
                        id=tc.get("id", ""),
                    )
                )

        logger.debug(
            "Loop done Content: %s  Reasoning: %s Tool Calls: %s",
            accumulated_content,
            accumulated_reasoning_content,
            len(tool_calls),
        )
        message = Message(
            role="assistant",
            message_id=generate_id(None),
            content=blocks,
            delta=False,
            reasoning=accumulated_reasoning_content,
            tools_calls=tool_calls,
            metadata=metadata,
        )
        yield message

    async def convert_streaming_response(  # type: ignore
        self,
        config: dict,
        node_name: str,
        response: Any,
        meta: dict | None = None,
    ) -> AsyncGenerator[Message]:
        """
        Convert a LiteLLM streaming or standard response to Message(s).

        Args:
            config (dict): Node configuration parameters.
            node_name (str): Name of the node processing the response.
            response (Any): The LiteLLM response object (stream or standard).
            meta (dict | None): Optional metadata for conversion.

        Yields:
            Message: Converted message(s) from the response.

        Raises:
            ImportError: If LiteLLM is not installed.
            Exception: If response type is unsupported.
        """
        if not HAS_LITELLM:
            raise ImportError("litellm is not installed. Please install it to use this converter.")

        if isinstance(response, CustomStreamWrapper):  # type: ignore[possibly-unbound]
            stream = cast(CustomStreamWrapper, response)
            async for event in self._handle_stream(
                config or {},
                node_name or "",
                stream,
                meta,
            ):
                yield event
        elif isinstance(response, ModelResponse):  # type: ignore[possibly-unbound]
            message = await self.convert_response(cast(ModelResponse, response))
            yield message
        else:
            raise Exception("Unsupported response type for LiteLLMConverter")

Attributes

state instance-attribute

state = state

Functions

__init__

__init__(state=None)

Initialize the converter.

Parameters:

Name	Type	Description	Default
state	AgentState | None	Optional agent state for context during conversion.	None

Source code in pyagenity/adapters/llm/base_converter.py

def __init__(self, state: AgentState | None = None) -> None:
    """
    Initialize the converter.

    Args:
        state (AgentState | None): Optional agent state for context during conversion.
    """
    self.state = state

convert_response async

convert_response(response)

Convert a LiteLLM ModelResponse to a Message.

Parameters:

Name	Type	Description	Default
response	ModelResponse	The LiteLLM model response object.	required

Returns:

Name	Type	Description
Message	Message	The converted message object.

Raises:

Type	Description
ImportError	If LiteLLM is not installed.

Source code in pyagenity/adapters/llm/litellm_converter.py

async def convert_response(self, response: ModelResponse) -> Message:
    """
    Convert a LiteLLM ModelResponse to a Message.

    Args:
        response (ModelResponse): The LiteLLM model response object.

    Returns:
        Message: The converted message object.

    Raises:
        ImportError: If LiteLLM is not installed.
    """
    if not HAS_LITELLM:
        raise ImportError("litellm is not installed. Please install it to use this converter.")

    data = response.model_dump()

    usages_data = data.get("usage", {})

    usages = TokenUsages(
        completion_tokens=usages_data.get("completion_tokens", 0),
        prompt_tokens=usages_data.get("prompt_tokens", 0),
        total_tokens=usages_data.get("total_tokens", 0),
        cache_creation_input_tokens=usages_data.get("cache_creation_input_tokens", 0),
        cache_read_input_tokens=usages_data.get("cache_read_input_tokens", 0),
        reasoning_tokens=usages_data.get("prompt_tokens_details", {}).get(
            "reasoning_tokens", 0
        ),
    )

    created_date = data.get("created", datetime.now())

    # Extract tool calls from response
    tools_calls = data.get("choices", [{}])[0].get("message", {}).get("tool_calls", []) or []

    logger.debug("Creating message from model response with id: %s", response.id)
    content = data.get("choices", [{}])[0].get("message", {}).get("content", "") or ""
    reasoning_content = (
        data.get("choices", [{}])[0].get("message", {}).get("reasoning_content", "") or ""
    )

    blocks = []
    if content:
        blocks.append(TextBlock(text=content))
    if reasoning_content:
        blocks.append(ReasoningBlock(summary=reasoning_content))
    final_tool_calls = []
    for tool_call in tools_calls:
        tool_id = tool_call.get("id", None)
        args = tool_call.get("function", {}).get("arguments", None)
        name = tool_call.get("function", {}).get("name", None)

        if not tool_id or not args or not name:
            continue

        blocks.append(
            ToolCallBlock(
                name=name,
                args=json.loads(args),
                id=tool_id,
            )
        )

        if hasattr(tool_call, "model_dump"):
            final_tool_calls.append(tool_call.model_dump())
        else:
            final_tool_calls.append(tool_call)

    return Message(
        message_id=generate_id(response.id),
        role="assistant",
        content=blocks,
        reasoning=reasoning_content,
        timestamp=created_date,
        metadata={
            "provider": "litellm",
            "model": data.get("model", ""),
            "finish_reason": data.get("choices", [{}])[0].get("finish_reason", "UNKNOWN"),
            "object": data.get("object", ""),
            "prompt_tokens_details": usages_data.get("prompt_tokens_details", {}),
            "completion_tokens_details": usages_data.get("completion_tokens_details", {}),
        },
        usages=usages,
        raw=data,
        tools_calls=final_tool_calls if final_tool_calls else None,
    )

convert_streaming_response async

convert_streaming_response(config, node_name, response, meta=None)

Convert a LiteLLM streaming or standard response to Message(s).

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters.	required
node_name	str	Name of the node processing the response.	required
response	Any	The LiteLLM response object (stream or standard).	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Name	Type	Description
Message	AsyncGenerator[Message]	Converted message(s) from the response.

Raises:

Type	Description
ImportError	If LiteLLM is not installed.
Exception	If response type is unsupported.

Source code in pyagenity/adapters/llm/litellm_converter.py

async def convert_streaming_response(  # type: ignore
    self,
    config: dict,
    node_name: str,
    response: Any,
    meta: dict | None = None,
) -> AsyncGenerator[Message]:
    """
    Convert a LiteLLM streaming or standard response to Message(s).

    Args:
        config (dict): Node configuration parameters.
        node_name (str): Name of the node processing the response.
        response (Any): The LiteLLM response object (stream or standard).
        meta (dict | None): Optional metadata for conversion.

    Yields:
        Message: Converted message(s) from the response.

    Raises:
        ImportError: If LiteLLM is not installed.
        Exception: If response type is unsupported.
    """
    if not HAS_LITELLM:
        raise ImportError("litellm is not installed. Please install it to use this converter.")

    if isinstance(response, CustomStreamWrapper):  # type: ignore[possibly-unbound]
        stream = cast(CustomStreamWrapper, response)
        async for event in self._handle_stream(
            config or {},
            node_name or "",
            stream,
            meta,
        ):
            yield event
    elif isinstance(response, ModelResponse):  # type: ignore[possibly-unbound]
        message = await self.convert_response(cast(ModelResponse, response))
        yield message
    else:
        raise Exception("Unsupported response type for LiteLLMConverter")

Functions

model_response_converter

Classes:

Name	Description
ModelResponseConverter	Wrap an LLM SDK call and normalize its output via a converter.

Classes

ModelResponseConverter

Wrap an LLM SDK call and normalize its output via a converter.

Supports sync/async invocation and streaming. Use invoke() for non-streaming calls and stream() for streaming calls.

Methods:

Name	Description
__init__	Initialize ModelResponseConverter.
invoke	Call the underlying function and convert a non-streaming response to Message.
stream	Call the underlying function and yield normalized streaming events and final Message.

Attributes:

Name	Type	Description
converter
response

Source code in pyagenity/adapters/llm/model_response_converter.py

class ModelResponseConverter:
    """Wrap an LLM SDK call and normalize its output via a converter.

    Supports sync/async invocation and streaming. Use `invoke()` for
    non-streaming calls and `stream()` for streaming calls.
    """

    def __init__(
        self,
        response: Any | Callable[..., Any],
        converter: BaseConverter | str,
    ) -> None:
        """
        Initialize ModelResponseConverter.

        Args:
            response (Any | Callable[..., Any]): The LLM response or a callable returning
                a response.
            converter (BaseConverter | str): Converter instance or string identifier
                (e.g., "litellm").

        Raises:
            ValueError: If the converter is not supported.
        """
        self.response = response

        if isinstance(converter, str) and converter == "litellm":
            from .litellm_converter import LiteLLMConverter

            self.converter = LiteLLMConverter()
        elif isinstance(converter, BaseConverter):
            self.converter = converter
        else:
            raise ValueError(f"Unsupported converter: {converter}")

    async def invoke(self) -> Message:
        """
        Call the underlying function and convert a non-streaming response to Message.

        Returns:
            Message: The normalized message from the LLM response.

        Raises:
            Exception: If the underlying function or converter fails.
        """
        if callable(self.response):
            if inspect.iscoroutinefunction(self.response):
                response = await self.response()
            else:
                response = self.response()
        else:
            response = self.response

        return await self.converter.convert_response(response)  # type: ignore

    async def stream(
        self,
        config: dict,
        node_name: str,
        meta: dict | None = None,
    ) -> AsyncGenerator[Message]:
        """
        Call the underlying function and yield normalized streaming events and final Message.

        Args:
            config (dict): Node configuration parameters for streaming.
            node_name (str): Name of the node processing the response.
            meta (dict | None): Optional metadata for conversion.

        Yields:
            Message: Normalized streaming message events from the LLM response.

        Raises:
            ValueError: If config is not provided.
            Exception: If the underlying function or converter fails.
        """
        if not config:
            raise ValueError("Config must be provided for streaming conversion")

        if callable(self.response):
            if inspect.iscoroutinefunction(self.response):
                response = await self.response()
            else:
                response = self.response()
        else:
            response = self.response

        async for item in self.converter.convert_streaming_response(  # type: ignore
            config,
            node_name=node_name,
            response=response,
            meta=meta,
        ):
            yield item

Attributes

converter instance-attribute

converter = LiteLLMConverter()

response instance-attribute

response = response

Functions

__init__

__init__(response, converter)

Initialize ModelResponseConverter.

Parameters:

Name	Type	Description	Default
response	Any | Callable[..., Any]	The LLM response or a callable returning a response.	required
converter	BaseConverter | str	Converter instance or string identifier (e.g., "litellm").	required

Raises:

Type	Description
ValueError	If the converter is not supported.

Source code in pyagenity/adapters/llm/model_response_converter.py

def __init__(
    self,
    response: Any | Callable[..., Any],
    converter: BaseConverter | str,
) -> None:
    """
    Initialize ModelResponseConverter.

    Args:
        response (Any | Callable[..., Any]): The LLM response or a callable returning
            a response.
        converter (BaseConverter | str): Converter instance or string identifier
            (e.g., "litellm").

    Raises:
        ValueError: If the converter is not supported.
    """
    self.response = response

    if isinstance(converter, str) and converter == "litellm":
        from .litellm_converter import LiteLLMConverter

        self.converter = LiteLLMConverter()
    elif isinstance(converter, BaseConverter):
        self.converter = converter
    else:
        raise ValueError(f"Unsupported converter: {converter}")

invoke async

invoke()

Call the underlying function and convert a non-streaming response to Message.

Returns:

Name	Type	Description
Message	Message	The normalized message from the LLM response.

Raises:

Type	Description
Exception	If the underlying function or converter fails.

Source code in pyagenity/adapters/llm/model_response_converter.py

async def invoke(self) -> Message:
    """
    Call the underlying function and convert a non-streaming response to Message.

    Returns:
        Message: The normalized message from the LLM response.

    Raises:
        Exception: If the underlying function or converter fails.
    """
    if callable(self.response):
        if inspect.iscoroutinefunction(self.response):
            response = await self.response()
        else:
            response = self.response()
    else:
        response = self.response

    return await self.converter.convert_response(response)  # type: ignore

stream async

stream(config, node_name, meta=None)

Call the underlying function and yield normalized streaming events and final Message.

Parameters:

Name	Type	Description	Default
config	dict	Node configuration parameters for streaming.	required
node_name	str	Name of the node processing the response.	required
meta	dict | None	Optional metadata for conversion.	None

Yields:

Name	Type	Description
Message	AsyncGenerator[Message]	Normalized streaming message events from the LLM response.

Raises:

Type	Description
ValueError	If config is not provided.
Exception	If the underlying function or converter fails.

Source code in pyagenity/adapters/llm/model_response_converter.py

async def stream(
    self,
    config: dict,
    node_name: str,
    meta: dict | None = None,
) -> AsyncGenerator[Message]:
    """
    Call the underlying function and yield normalized streaming events and final Message.

    Args:
        config (dict): Node configuration parameters for streaming.
        node_name (str): Name of the node processing the response.
        meta (dict | None): Optional metadata for conversion.

    Yields:
        Message: Normalized streaming message events from the LLM response.

    Raises:
        ValueError: If config is not provided.
        Exception: If the underlying function or converter fails.
    """
    if not config:
        raise ValueError("Config must be provided for streaming conversion")

    if callable(self.response):
        if inspect.iscoroutinefunction(self.response):
            response = await self.response()
        else:
            response = self.response()
    else:
        response = self.response

    async for item in self.converter.convert_streaming_response(  # type: ignore
        config,
        node_name=node_name,
        response=response,
        meta=meta,
    ):
        yield item

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,
        },
    }