Skip to content

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