resource_providers

Class info

Classes

Name	Children	Inherits
CallableResourceProvider llmling_agent.resource_providers.callable_provider Wraps callables as a resource provider.		ResourceProvider
ResourceProvider llmling_agent.resource_providers.base Base class for resource providers.	CallableResourceProvider StaticResourceProvider MCPManager RuntimeResourceProvider CapabilitiesResourceProvider	object

🛈 DocStrings

Resource provider implementations.

CallableResourceProvider

Bases: ResourceProvider

Wraps callables as a resource provider.

Handles both sync and async functions transparently.

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

class CallableResourceProvider(ResourceProvider):
    """Wraps callables as a resource provider.

    Handles both sync and async functions transparently.
    """

    def __init__(
        self,
        name: str = "callable",
        tool_callable: ResourceCallable[list[ToolInfo]] | None = None,
        prompt_callable: ResourceCallable[list[BasePrompt]] | None = None,
        resource_callable: ResourceCallable[list[ResourceInfo]] | None = None,
    ):
        """Initialize provider with optional callables.

        Args:
            name: Name of the provider
            tool_callable: Function providing tools
            prompt_callable: Function providing prompts
            resource_callable: Function providing resources

        Each callable can be sync or async.
        """
        super().__init__(name=name)
        self.tool_callable = tool_callable
        self.prompt_callable = prompt_callable
        self.resource_callable = resource_callable

    async def _call_provider[T](
        self,
        provider: ResourceCallable[T] | None,
        default: T,
    ) -> T:
        """Helper to handle sync/async provider calls."""
        if not provider:
            return default
        return await execute(provider)

    async def get_tools(self) -> list[ToolInfo]:
        """Get tools from callable if provided."""
        return await self._call_provider(self.tool_callable, [])

    async def get_prompts(self) -> list[BasePrompt]:
        """Get prompts from callable if provided."""
        return await self._call_provider(self.prompt_callable, [])

    async def get_resources(self) -> list[ResourceInfo]:
        """Get resources from callable if provided."""
        return await self._call_provider(self.resource_callable, [])

__init__

__init__(
    name: str = "callable",
    tool_callable: ResourceCallable[list[ToolInfo]] | None = None,
    prompt_callable: ResourceCallable[list[BasePrompt]] | None = None,
    resource_callable: ResourceCallable[list[ResourceInfo]] | None = None,
)

Initialize provider with optional callables.

Parameters:

Name	Type	Description	Default
name	str	Name of the provider	'callable'
tool_callable	ResourceCallable[list[ToolInfo]] | None	Function providing tools	None
prompt_callable	ResourceCallable[list[BasePrompt]] | None	Function providing prompts	None
resource_callable	ResourceCallable[list[ResourceInfo]] | None	Function providing resources	None

Each callable can be sync or async.

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

def __init__(
    self,
    name: str = "callable",
    tool_callable: ResourceCallable[list[ToolInfo]] | None = None,
    prompt_callable: ResourceCallable[list[BasePrompt]] | None = None,
    resource_callable: ResourceCallable[list[ResourceInfo]] | None = None,
):
    """Initialize provider with optional callables.

    Args:
        name: Name of the provider
        tool_callable: Function providing tools
        prompt_callable: Function providing prompts
        resource_callable: Function providing resources

    Each callable can be sync or async.
    """
    super().__init__(name=name)
    self.tool_callable = tool_callable
    self.prompt_callable = prompt_callable
    self.resource_callable = resource_callable

_call_provider async

_call_provider(provider: ResourceCallable[T] | None, default: T) -> T

Helper to handle sync/async provider calls.

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

async def _call_provider[T](
    self,
    provider: ResourceCallable[T] | None,
    default: T,
) -> T:
    """Helper to handle sync/async provider calls."""
    if not provider:
        return default
    return await execute(provider)

get_prompts async

get_prompts() -> list[BasePrompt]

Get prompts from callable if provided.

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

async def get_prompts(self) -> list[BasePrompt]:
    """Get prompts from callable if provided."""
    return await self._call_provider(self.prompt_callable, [])

get_resources async

get_resources() -> list[ResourceInfo]

Get resources from callable if provided.

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

async def get_resources(self) -> list[ResourceInfo]:
    """Get resources from callable if provided."""
    return await self._call_provider(self.resource_callable, [])

get_tools async

get_tools() -> list[ToolInfo]

Get tools from callable if provided.

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

async def get_tools(self) -> list[ToolInfo]:
    """Get tools from callable if provided."""
    return await self._call_provider(self.tool_callable, [])

ResourceProvider

Base class for resource providers.

Provides tools, prompts, and other resources to agents. Default implementations return empty lists - override as needed.

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

class ResourceProvider:
    """Base class for resource providers.

    Provides tools, prompts, and other resources to agents.
    Default implementations return empty lists - override as needed.
    """

    def __init__(self, name: str, owner: str | None = None) -> None:
        """Initialize the resource provider."""
        self.name = name
        self.owner = owner

    async def __aenter__(self) -> Self:
        """Async context entry if required."""
        return self

    async def __aexit__(self, *exc: object) -> None:
        """Async context cleanup if required."""

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name!r})"

    @property
    def requires_async(self) -> bool:
        return False

    async def get_tools(self) -> list[ToolInfo]:
        """Get available tools. Override to provide tools."""
        return []

    async def get_prompts(self) -> list[BasePrompt]:
        """Get available prompts. Override to provide prompts."""
        return []

    async def get_resources(self) -> list[ResourceInfo]:
        """Get available resources. Override to provide resources."""
        return []

    async def get_formatted_prompt(
        self, name: str, arguments: dict[str, str] | None = None
    ) -> ChatMessage[str]:
        """Get a prompt formatted with arguments.

        Args:
            name: Name of the prompt to format
            arguments: Optional arguments for prompt formatting

        Returns:
            Single chat message with merged content

        Raises:
            KeyError: If prompt not found
            ValueError: If formatting fails
        """
        from llmling_agent.messaging.messages import ChatMessage

        prompts = await self.get_prompts()
        prompt = next((p for p in prompts if p.name == name), None)
        if not prompt:
            msg = f"Prompt {name!r} not found"
            raise KeyError(msg)

        messages = await prompt.format(arguments or {})
        if not messages:
            msg = f"Prompt {name!r} produced no messages"
            raise ValueError(msg)

        # Use role from first message (usually system)
        role = messages[0].role
        # Merge all message contents
        content = "\n\n".join(msg.get_text_content() for msg in messages)

        return ChatMessage(
            content=content,
            role=role,  # type: ignore
            name=self.name,
            metadata={
                "prompt_name": name,
                "arguments": arguments or {},  # type: ignore
            },
        )

__aenter__ async

__aenter__() -> Self

Async context entry if required.

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

async def __aenter__(self) -> Self:
    """Async context entry if required."""
    return self

__aexit__ async

__aexit__(*exc: object) -> None

Async context cleanup if required.

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

async def __aexit__(self, *exc: object) -> None:
    """Async context cleanup if required."""

__init__

__init__(name: str, owner: str | None = None) -> None

Initialize the resource provider.

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

def __init__(self, name: str, owner: str | None = None) -> None:
    """Initialize the resource provider."""
    self.name = name
    self.owner = owner

get_formatted_prompt async

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

Get a prompt formatted with arguments.

Parameters:

Name	Type	Description	Default
name	str	Name of the prompt to format	required
arguments	dict[str, str] | None	Optional arguments for prompt formatting	None

Returns:

Type	Description
ChatMessage[str]	Single chat message with merged content

Raises:

Type	Description
KeyError	If prompt not found
ValueError	If formatting fails

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

async def get_formatted_prompt(
    self, name: str, arguments: dict[str, str] | None = None
) -> ChatMessage[str]:
    """Get a prompt formatted with arguments.

    Args:
        name: Name of the prompt to format
        arguments: Optional arguments for prompt formatting

    Returns:
        Single chat message with merged content

    Raises:
        KeyError: If prompt not found
        ValueError: If formatting fails
    """
    from llmling_agent.messaging.messages import ChatMessage

    prompts = await self.get_prompts()
    prompt = next((p for p in prompts if p.name == name), None)
    if not prompt:
        msg = f"Prompt {name!r} not found"
        raise KeyError(msg)

    messages = await prompt.format(arguments or {})
    if not messages:
        msg = f"Prompt {name!r} produced no messages"
        raise ValueError(msg)

    # Use role from first message (usually system)
    role = messages[0].role
    # Merge all message contents
    content = "\n\n".join(msg.get_text_content() for msg in messages)

    return ChatMessage(
        content=content,
        role=role,  # type: ignore
        name=self.name,
        metadata={
            "prompt_name": name,
            "arguments": arguments or {},  # type: ignore
        },
    )

get_prompts async

get_prompts() -> list[BasePrompt]

Get available prompts. Override to provide prompts.

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

async def get_prompts(self) -> list[BasePrompt]:
    """Get available prompts. Override to provide prompts."""
    return []

get_resources async

get_resources() -> list[ResourceInfo]

Get available resources. Override to provide resources.

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

async def get_resources(self) -> list[ResourceInfo]:
    """Get available resources. Override to provide resources."""
    return []

get_tools async

get_tools() -> list[ToolInfo]

Get available tools. Override to provide tools.

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

async def get_tools(self) -> list[ToolInfo]:
    """Get available tools. Override to provide tools."""
    return []