aggregating

Class info

Classes

Name	Children	Inherits
AggregatingResourceProvider llmling_agent.resource_providers.aggregating Provider that combines resources from multiple providers.		ResourceProvider
ResourceProvider llmling_agent.resource_providers.base Base class for resource providers.	CallableResourceProvider StaticResourceProvider MCPManager RuntimeResourceProvider CapabilitiesResourceProvider	object

🛈 DocStrings

AggregatingResourceProvider

Bases: ResourceProvider

Provider that combines resources from multiple providers.

Source code in src/llmling_agent/resource_providers/aggregating.py

class AggregatingResourceProvider(ResourceProvider):
    """Provider that combines resources from multiple providers."""

    def __init__(self, providers: Sequence[ResourceProvider], name: str = "aggregating"):
        """Initialize provider with list of providers to aggregate.

        Args:
            providers: Resource providers to aggregate
            name: Name for this provider
        """
        super().__init__(name=name)
        self.providers = list(providers)

    async def get_tools(self) -> list[ToolInfo]:
        """Get tools from all providers."""
        tools: list[ToolInfo] = []
        for provider in self.providers:
            tools.extend(await provider.get_tools())
        return tools

    async def get_prompts(self) -> list[BasePrompt]:
        """Get prompts from all providers."""
        prompts: list[BasePrompt] = []
        for provider in self.providers:
            prompts.extend(await provider.get_prompts())
        return prompts

    async def get_resources(self) -> list[ResourceInfo]:
        """Get resources from all providers."""
        resources: list[ResourceInfo] = []
        for provider in self.providers:
            resources.extend(await provider.get_resources())
        return resources

    async def get_formatted_prompt(
        self, name: str, arguments: dict[str, str] | None = None
    ) -> ChatMessage[str]:
        """Try to get prompt from first provider that has it."""
        for provider in self.providers:
            try:
                return await provider.get_formatted_prompt(name, arguments)
            except KeyError:
                continue
        msg = f"Prompt {name!r} not found in any provider"
        raise KeyError(msg)

    @property
    def requires_async(self) -> bool:
        return any(p.requires_async for p in self.providers)

__init__

__init__(providers: Sequence[ResourceProvider], name: str = 'aggregating')

Initialize provider with list of providers to aggregate.

Parameters:

Name	Type	Description	Default
providers	Sequence[ResourceProvider]	Resource providers to aggregate	required
name	str	Name for this provider	'aggregating'

Source code in src/llmling_agent/resource_providers/aggregating.py

def __init__(self, providers: Sequence[ResourceProvider], name: str = "aggregating"):
    """Initialize provider with list of providers to aggregate.

    Args:
        providers: Resource providers to aggregate
        name: Name for this provider
    """
    super().__init__(name=name)
    self.providers = list(providers)

get_formatted_prompt async

get_formatted_prompt(
    name: str, arguments: dict[str, str] | None = None
) -> ChatMessage[str]

Try to get prompt from first provider that has it.

Source code in src/llmling_agent/resource_providers/aggregating.py

async def get_formatted_prompt(
    self, name: str, arguments: dict[str, str] | None = None
) -> ChatMessage[str]:
    """Try to get prompt from first provider that has it."""
    for provider in self.providers:
        try:
            return await provider.get_formatted_prompt(name, arguments)
        except KeyError:
            continue
    msg = f"Prompt {name!r} not found in any provider"
    raise KeyError(msg)

get_prompts async

get_prompts() -> list[BasePrompt]

Get prompts from all providers.

Source code in src/llmling_agent/resource_providers/aggregating.py

async def get_prompts(self) -> list[BasePrompt]:
    """Get prompts from all providers."""
    prompts: list[BasePrompt] = []
    for provider in self.providers:
        prompts.extend(await provider.get_prompts())
    return prompts

get_resources async

get_resources() -> list[ResourceInfo]

Get resources from all providers.

Source code in src/llmling_agent/resource_providers/aggregating.py

async def get_resources(self) -> list[ResourceInfo]:
    """Get resources from all providers."""
    resources: list[ResourceInfo] = []
    for provider in self.providers:
        resources.extend(await provider.get_resources())
    return resources

get_tools async

get_tools() -> list[ToolInfo]

Get tools from all providers.

Source code in src/llmling_agent/resource_providers/aggregating.py

async def get_tools(self) -> list[ToolInfo]:
    """Get tools from all providers."""
    tools: list[ToolInfo] = []
    for provider in self.providers:
        tools.extend(await provider.get_tools())
    return tools