Agents

ragbits.agents.AgentOptions

Bases: Options, Generic[LLMClientOptionsT]

Options for the agent run.

model_config class-attribute instance-attribute

model_config = ConfigDict(extra='allow', arbitrary_types_allowed=True)

llm_options class-attribute instance-attribute

llm_options: LLMClientOptionsT | None | NotGiven = NOT_GIVEN

The options for the LLM.

max_turns class-attribute instance-attribute

max_turns: int | None | NotGiven = NOT_GIVEN

The maximum number of turns the agent can take, if NOT_GIVEN, it defaults to 10, if None, agent will run forever

dict

dict() -> dict[str, Any]

Creates a dictionary representation of the Options instance. If a value is None, it will be replaced with a provider-specific not-given sentinel.

RETURNS	DESCRIPTION
dict[str, Any]	A dictionary representation of the Options instance.

Source code in packages/ragbits-core/src/ragbits/core/options.py

def dict(self) -> dict[str, Any]:  # type: ignore # mypy complains about overriding BaseModel.dict
    """
    Creates a dictionary representation of the Options instance.
    If a value is None, it will be replaced with a provider-specific not-given sentinel.

    Returns:
        A dictionary representation of the Options instance.
    """
    options = self.model_dump()

    return {
        key: self._not_given if value is None or isinstance(value, NotGiven) else value
        for key, value in options.items()
    }

ragbits.agents.Agent

Agent(llm: LLM[LLMClientOptionsT], prompt: str | type[Prompt[PromptInputT, PromptOutputT]] | Prompt[PromptInputT, PromptOutputT] | None = None, *, history: ChatFormat | None = None, keep_history: bool = False, tools: list[Callable] | None = None, mcp_servers: list[MCPServer] | None = None, default_options: AgentOptions[LLMClientOptionsT] | None = None)

Bases: ConfigurableComponent[AgentOptions[LLMClientOptionsT]], Generic[LLMClientOptionsT, PromptInputT, PromptOutputT]

Agent class that orchestrates the LLM and the prompt, and can call tools.

Current implementation is highly experimental, and the API is subject to change.

Initialize the agent instance.

PARAMETER	DESCRIPTION
llm	The LLM to run the agent. TYPE: LLM[LLMClientOptionsT]
prompt	The prompt for the agent. Can be: - str: A string prompt that will be used as system message when combined with string input, or as the user message when no input is provided during run(). - type[Prompt]: A structured prompt class that will be instantiated with the input. - Prompt: Already instantiated prompt instance - None: No predefined prompt. The input provided to run() will be used as the complete prompt. TYPE: str | type[Prompt[PromptInputT, PromptOutputT]] | Prompt[PromptInputT, PromptOutputT] | None DEFAULT: None
history	The history of the agent. TYPE: ChatFormat | None DEFAULT: None
keep_history	Whether to keep the history of the agent. TYPE: bool DEFAULT: False
tools	The tools available to the agent. TYPE: list[Callable] | None DEFAULT: None
mcp_servers	The MCP servers available to the agent. TYPE: list[MCPServer] | None DEFAULT: None
default_options	The default options for the agent run. TYPE: AgentOptions[LLMClientOptionsT] | None DEFAULT: None

Source code in packages/ragbits-agents/src/ragbits/agents/_main.py

def __init__(
    self,
    llm: LLM[LLMClientOptionsT],
    prompt: str | type[Prompt[PromptInputT, PromptOutputT]] | Prompt[PromptInputT, PromptOutputT] | None = None,
    *,
    history: ChatFormat | None = None,
    keep_history: bool = False,
    tools: list[Callable] | None = None,
    mcp_servers: list[MCPServer] | None = None,
    default_options: AgentOptions[LLMClientOptionsT] | None = None,
) -> None:
    """
    Initialize the agent instance.

    Args:
        llm: The LLM to run the agent.
        prompt: The prompt for the agent. Can be:
            - str: A string prompt that will be used as system message when combined with string input,
                or as the user message when no input is provided during run().
            - type[Prompt]: A structured prompt class that will be instantiated with the input.
            - Prompt: Already instantiated prompt instance
            - None: No predefined prompt. The input provided to run() will be used as the complete prompt.
        history: The history of the agent.
        keep_history: Whether to keep the history of the agent.
        tools: The tools available to the agent.
        mcp_servers: The MCP servers available to the agent.
        default_options: The default options for the agent run.
    """
    super().__init__(default_options)
    self.llm = llm
    self.prompt = prompt
    self.tools = [Tool.from_callable(tool) for tool in tools or []]
    self.mcp_servers = mcp_servers or []
    self.history = history or []
    self.keep_history = keep_history

default_options instance-attribute

default_options: OptionsT = default_options or options_cls()

options_cls class-attribute instance-attribute

options_cls: type[AgentOptions] = AgentOptions

default_module class-attribute

default_module: ModuleType | None = agents

configuration_key class-attribute

configuration_key: str = 'agent'

llm instance-attribute

llm = llm

prompt instance-attribute

prompt = prompt

tools instance-attribute

tools = [from_callable(tool) for tool in tools or []]

mcp_servers instance-attribute

mcp_servers = mcp_servers or []

history instance-attribute

history = history or []

keep_history instance-attribute

keep_history = keep_history

subclass_from_config classmethod

subclass_from_config(config: ObjectConstructionConfig) -> Self

Initializes the class with the provided configuration. May return a subclass of the class, if requested by the configuration.

PARAMETER	DESCRIPTION
config	A model containing configuration details for the class. TYPE: ObjectConstructionConfig

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

RAISES	DESCRIPTION
InvalidConfigError	The class can't be found or is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_config(cls, config: ObjectConstructionConfig) -> Self:
    """
    Initializes the class with the provided configuration. May return a subclass of the class,
    if requested by the configuration.

    Args:
        config: A model containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.

    Raises:
        InvalidConfigError: The class can't be found or is not a subclass of the current class.
    """
    subclass = import_by_path(config.type, cls.default_module)
    if not issubclass(subclass, cls):
        raise InvalidConfigError(f"{subclass} is not a subclass of {cls}")

    return subclass.from_config(config.config)

subclass_from_factory classmethod

subclass_from_factory(factory_path: str) -> Self

Creates the class using the provided factory function. May return a subclass of the class, if requested by the factory. Supports both synchronous and asynchronous factory functions.

PARAMETER	DESCRIPTION
factory_path	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided factory function.

RAISES	DESCRIPTION
InvalidConfigError	The factory can't be found or the object returned is not a subclass of the current class.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def subclass_from_factory(cls, factory_path: str) -> Self:
    """
    Creates the class using the provided factory function. May return a subclass of the class,
    if requested by the factory. Supports both synchronous and asynchronous factory functions.

    Args:
        factory_path: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".

    Returns:
        An instance of the class initialized with the provided factory function.

    Raises:
        InvalidConfigError: The factory can't be found or the object returned
            is not a subclass of the current class.
    """
    factory = import_by_path(factory_path, cls.default_module)

    if asyncio.iscoroutinefunction(factory):
        try:
            loop = asyncio.get_running_loop()
            obj = asyncio.run_coroutine_threadsafe(factory, loop).result()
        except RuntimeError:
            obj = asyncio.run(factory())
    else:
        obj = factory()

    if not isinstance(obj, cls):
        raise InvalidConfigError(f"The object returned by factory {factory_path} is not an instance of {cls}")

    return obj

preferred_subclass classmethod

preferred_subclass(config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None) -> Self

Tries to create an instance by looking at project's component preferences, either from YAML or from the factory. Takes optional overrides for both, which takes a higher precedence.

PARAMETER	DESCRIPTION
config	The CoreConfig instance containing preferred factory and configuration details. TYPE: CoreConfig
factory_path_override	A string representing the path to the factory function in the format of "module.submodule:factory_name". TYPE: str | None DEFAULT: None
yaml_path_override	A string representing the path to the YAML file containing the Ragstack instance configuration. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
InvalidConfigError	If the default factory or configuration can't be found.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def preferred_subclass(
    cls, config: CoreConfig, factory_path_override: str | None = None, yaml_path_override: Path | None = None
) -> Self:
    """
    Tries to create an instance by looking at project's component preferences, either from YAML
    or from the factory. Takes optional overrides for both, which takes a higher precedence.

    Args:
        config: The CoreConfig instance containing preferred factory and configuration details.
        factory_path_override: A string representing the path to the factory function
            in the format of "module.submodule:factory_name".
        yaml_path_override: A string representing the path to the YAML file containing
            the Ragstack instance configuration.

    Raises:
        InvalidConfigError: If the default factory or configuration can't be found.
    """
    if yaml_path_override:
        preferences = get_config_from_yaml(yaml_path_override)
        if type_config := preferences.get(cls.configuration_key):
            return cls.subclass_from_config(ObjectConstructionConfig.model_validate(type_config))

    if factory_path_override:
        return cls.subclass_from_factory(factory_path_override)

    if preferred_factory := config.component_preference_factories.get(cls.configuration_key):
        return cls.subclass_from_factory(preferred_factory)

    if preferred_config := config.preferred_instances_config.get(cls.configuration_key):
        return cls.subclass_from_config(ObjectConstructionConfig.model_validate(preferred_config))

    raise NoPreferredConfigError(f"Could not find preferred factory or configuration for {cls.configuration_key}")

from_config classmethod

from_config(config: dict[str, Any]) -> Self

Initializes the class with the provided configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing configuration details for the class. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Self	An instance of the class initialized with the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/utils/config_handling.py

@classmethod
def from_config(cls, config: dict[str, Any]) -> Self:
    """
    Initializes the class with the provided configuration.

    Args:
        config: A dictionary containing configuration details for the class.

    Returns:
        An instance of the class initialized with the provided configuration.
    """
    default_options = config.pop("default_options", None)
    options = cls.options_cls(**default_options) if default_options else None
    return cls(**config, default_options=options)

run async

run(input: str | PromptInputT | None = None, options: AgentOptions[LLMClientOptionsT] | None = None) -> AgentResult[PromptOutputT]

Run the agent. The method is experimental, inputs and outputs may change in the future.

PARAMETER	DESCRIPTION
input	The input for the agent run. Can be: - str: A string input that will be used as user message. - PromptInputT: Structured input for use with structured prompt classes. - None: No input. Only valid when a string prompt was provided during initialization. TYPE: str | PromptInputT | None DEFAULT: None
options	The options for the agent run. TYPE: AgentOptions[LLMClientOptionsT] | None DEFAULT: None

RETURNS	DESCRIPTION
AgentResult[PromptOutputT]	The result of the agent run.

RAISES	DESCRIPTION
AgentToolDuplicateError	If the tool names are duplicated.
AgentToolNotSupportedError	If the selected tool type is not supported.
AgentToolNotAvailableError	If the selected tool is not available.
AgentInvalidPromptInputError	If the prompt/input combination is invalid.
AgentMaxTurnsExceededError	If the maximum number of turns is exceeded.

Source code in packages/ragbits-agents/src/ragbits/agents/_main.py

async def run(
    self, input: str | PromptInputT | None = None, options: AgentOptions[LLMClientOptionsT] | None = None
) -> AgentResult[PromptOutputT]:
    """
    Run the agent. The method is experimental, inputs and outputs may change in the future.

    Args:
        input: The input for the agent run. Can be:
            - str: A string input that will be used as user message.
            - PromptInputT: Structured input for use with structured prompt classes.
            - None: No input. Only valid when a string prompt was provided during initialization.
        options: The options for the agent run.

    Returns:
        The result of the agent run.

    Raises:
        AgentToolDuplicateError: If the tool names are duplicated.
        AgentToolNotSupportedError: If the selected tool type is not supported.
        AgentToolNotAvailableError: If the selected tool is not available.
        AgentInvalidPromptInputError: If the prompt/input combination is invalid.
        AgentMaxTurnsExceededError: If the maximum number of turns is exceeded.
    """
    input = cast(PromptInputT, input)
    merged_options = (self.default_options | options) if options else self.default_options
    llm_options = merged_options.llm_options or None

    prompt_with_history = self._get_prompt_with_history(input)
    tools_mapping = await self._get_all_tools()
    tool_calls = []

    turn_count = 0
    max_turns = merged_options.max_turns
    max_turns = 10 if max_turns is NOT_GIVEN else max_turns
    with trace(input=input, options=merged_options) as outputs:
        while not max_turns or turn_count < max_turns:
            response = cast(
                LLMResponseWithMetadata[PromptOutputT],
                await self.llm.generate_with_metadata(
                    prompt=prompt_with_history,
                    tools=[tool.to_function_schema() for tool in tools_mapping.values()],
                    options=llm_options,
                ),
            )
            if not response.tool_calls:
                break

            for tool_call in response.tool_calls:
                result = await self._execute_tool(tool_call=tool_call, tools_mapping=tools_mapping)
                tool_calls.append(result)

                prompt_with_history = prompt_with_history.add_tool_use_message(**result.__dict__)

            turn_count += 1
        else:
            raise AgentMaxTurnsExceededError(cast(int, max_turns))

        outputs.result = {
            "content": response.content,
            "metadata": response.metadata,
            "tool_calls": tool_calls or None,
        }

        prompt_with_history = prompt_with_history.add_assistant_message(response.content)

        if self.keep_history:
            self.history = prompt_with_history.chat

        return AgentResult(
            content=response.content,
            metadata=response.metadata,
            tool_calls=tool_calls or None,
            history=prompt_with_history.chat,
        )

run_streaming

run_streaming(input: str | PromptInputT | None = None, options: AgentOptions[LLMClientOptionsT] | None = None) -> AgentResultStreaming

This method returns an AgentResultStreaming object that can be asynchronously iterated over. After the loop completes, all items are available under the same names as in AgentResult class.

PARAMETER	DESCRIPTION
input	The input for the agent run. TYPE: str | PromptInputT | None DEFAULT: None
options	The options for the agent run. TYPE: AgentOptions[LLMClientOptionsT] | None DEFAULT: None

RETURNS	DESCRIPTION
AgentResultStreaming	A StreamingResult object for iteration and collection.

RAISES	DESCRIPTION
AgentToolDuplicateError	If the tool names are duplicated.
AgentToolNotSupportedError	If the selected tool type is not supported.
AgentToolNotAvailableError	If the selected tool is not available.
AgentInvalidPromptInputError	If the prompt/input combination is invalid.
AgentMaxTurnsExceededError	If the maximum number of turns is exceeded.

Source code in packages/ragbits-agents/src/ragbits/agents/_main.py

def run_streaming(
    self, input: str | PromptInputT | None = None, options: AgentOptions[LLMClientOptionsT] | None = None
) -> AgentResultStreaming:
    """
    This method returns an `AgentResultStreaming` object that can be asynchronously
    iterated over. After the loop completes, all items are available under the same names as in AgentResult class.

    Args:
        input: The input for the agent run.
        options: The options for the agent run.

    Returns:
        A `StreamingResult` object for iteration and collection.

    Raises:
        AgentToolDuplicateError: If the tool names are duplicated.
        AgentToolNotSupportedError: If the selected tool type is not supported.
        AgentToolNotAvailableError: If the selected tool is not available.
        AgentInvalidPromptInputError: If the prompt/input combination is invalid.
        AgentMaxTurnsExceededError: If the maximum number of turns is exceeded.
    """
    generator = self._stream_internal(input, options)
    return AgentResultStreaming(generator)

get_agent_card async

get_agent_card(name: str, description: str, version: str = '0.0.0', host: str = '127.0.0.1', port: int = 8000, protocol: str = 'http', default_input_modes: list[str] | None = None, default_output_modes: list[str] | None = None, capabilities: AgentCapabilities | None = None, skills: list[AgentSkill] | None = None) -> AgentCard

Create an AgentCard that encapsulates metadata about the agent, such as its name, version, description, network location, supported input/output modes, capabilities, and skills.

PARAMETER	DESCRIPTION
name	Human-readable name of the agent. TYPE: str
description	A brief description of the agent. TYPE: str
version	Version string of the agent. Defaults to "0.0.0". TYPE: str DEFAULT: '0.0.0'
host	Hostname or IP where the agent will be served. Defaults to "0.0.0.0". TYPE: str DEFAULT: '127.0.0.1'
port	Port number on which the agent listens. Defaults to 8000. TYPE: int DEFAULT: 8000
protocol	URL scheme (e.g. "http" or "https"). Defaults to "http". TYPE: str DEFAULT: 'http'
default_input_modes	List of input content modes supported by the agent. Defaults to ["text"]. TYPE: list[str] | None DEFAULT: None
default_output_modes	List of output content modes supported. Defaults to ["text"]. TYPE: list[str] | None DEFAULT: None
capabilities	Agent capabilities; if None, defaults to empty capabilities. TYPE: AgentCapabilities | None DEFAULT: None
skills	List of AgentSkill objects representing the agent's skills. If None, attempts to extract skills from the agent's registered tools. TYPE: list[AgentSkill] | None DEFAULT: None

RETURNS	DESCRIPTION
AgentCard	An A2A-compliant agent descriptor including URL and capabilities.

Source code in packages/ragbits-agents/src/ragbits/agents/_main.py

@requires_dependencies(["a2a.types"], "a2a")
async def get_agent_card(
    self,
    name: str,
    description: str,
    version: str = "0.0.0",
    host: str = "127.0.0.1",
    port: int = 8000,
    protocol: str = "http",
    default_input_modes: list[str] | None = None,
    default_output_modes: list[str] | None = None,
    capabilities: "AgentCapabilities | None" = None,
    skills: list["AgentSkill"] | None = None,
) -> "AgentCard":
    """
    Create an AgentCard that encapsulates metadata about the agent,
    such as its name, version, description, network location, supported input/output modes,
    capabilities, and skills.

    Args:
        name: Human-readable name of the agent.
        description: A brief description of the agent.
        version: Version string of the agent. Defaults to "0.0.0".
        host: Hostname or IP where the agent will be served. Defaults to "0.0.0.0".
        port: Port number on which the agent listens. Defaults to 8000.
        protocol: URL scheme (e.g. "http" or "https"). Defaults to "http".
        default_input_modes: List of input content modes supported by the agent. Defaults to ["text"].
        default_output_modes: List of output content modes supported. Defaults to ["text"].
        capabilities: Agent capabilities; if None, defaults to empty capabilities.
        skills: List of AgentSkill objects representing the agent's skills.
            If None, attempts to extract skills from the agent's registered tools.

    Returns:
        An A2A-compliant agent descriptor including URL and capabilities.
    """
    return AgentCard(
        name=name,
        version=version,
        description=description,
        url=f"{protocol}://{host}:{port}",
        defaultInputModes=default_input_modes or ["text"],
        defaultOutputModes=default_output_modes or ["text"],
        skills=skills or await self._extract_agent_skills(),
        capabilities=capabilities or AgentCapabilities(),
    )

ragbits.agents.AgentResult dataclass

AgentResult(content: PromptOutputT, metadata: dict, history: ChatFormat, tool_calls: list[ToolCallResult] | None = None)

Bases: Generic[PromptOutputT]

Result of the agent run.

content instance-attribute

content: PromptOutputT

The output content of the agent.

metadata instance-attribute

metadata: dict

The additional data returned by the agent.

history instance-attribute

history: ChatFormat

The history of the agent.

tool_calls class-attribute instance-attribute

tool_calls: list[ToolCallResult] | None = None

Tool calls run by the agent.

ragbits.agents.AgentResultStreaming

AgentResultStreaming(generator: AsyncGenerator[str | ToolCall | ToolCallResult | SimpleNamespace | BasePrompt])

Bases: AsyncIterator[str | ToolCall | ToolCallResult]

An async iterator that will collect all yielded items by LLM.generate_streaming(). This object is returned by run_streaming. It can be used in an async for loop to process items as they arrive. After the loop completes, all items are available under the same names as in AgentResult class.

Source code in packages/ragbits-agents/src/ragbits/agents/_main.py

def __init__(self, generator: AsyncGenerator[str | ToolCall | ToolCallResult | SimpleNamespace | BasePrompt]):
    self._generator = generator
    self.content: str = ""
    self.tool_calls: list[ToolCallResult] | None = None
    self.metadata: dict = {}
    self.history: ChatFormat

content instance-attribute

content: str = ''

tool_calls instance-attribute

tool_calls: list[ToolCallResult] | None = None

metadata instance-attribute

metadata: dict = {}

history instance-attribute

history: ChatFormat

ragbits.agents.a2a.server.create_agent_server

create_agent_server(agent: Agent, agent_card: AgentCard, input_model: type[BaseModel]) -> Server

Create a Uvicorn server instance that serves the specified agent over HTTP.

The server's host and port are extracted from the URL in the given agent_card.

PARAMETER	DESCRIPTION
agent	The Ragbits Agent instance to serve. TYPE: Agent
agent_card	Metadata for the agent, including its URL. TYPE: AgentCard
input_model	A Pydantic model class used to validate incoming request data. TYPE: type[BaseModel]

RETURNS	DESCRIPTION
Server	A configured uvicorn.Server instance ready to be started.

RAISES	DESCRIPTION
ValueError	If the URL in agent_card does not contain a valid hostname or port.

Source code in packages/ragbits-agents/src/ragbits/agents/a2a/server.py

def create_agent_server(
    agent: Agent,
    agent_card: "AgentCard",
    input_model: type[BaseModel],
) -> "uvicorn.Server":
    """
    Create a Uvicorn server instance that serves the specified agent over HTTP.

    The server's host and port are extracted from the URL in the given agent_card.

    Args:
        agent: The Ragbits Agent instance to serve.
        agent_card: Metadata for the agent, including its URL.
        input_model: A Pydantic model class used to validate incoming request data.

    Returns:
        A configured uvicorn.Server instance ready to be started.

    Raises:
        ValueError: If the URL in agent_card does not contain a valid hostname or port.
    """
    app = create_agent_app(agent=agent, agent_card=agent_card, input_model=input_model)
    url = urlparse(agent_card.url)

    if not url.hostname:
        raise ValueError(f"Could not parse hostname from URL: {agent_card.url}")
    if not url.port:
        raise ValueError(f"Could not parse port from URL: {agent_card.url}")

    config = uvicorn.Config(app=app, host=url.hostname, port=url.port)
    server = uvicorn.Server(config=config)

    return server