Skip to content

provider

Class info

Classes

Name Children Inherits
AgentContext
llmling_agent.agent.context
Runtime context for agent execution.
    AggregatingResourceProvider
    llmling_agent.resource_providers.aggregating
    Provider that combines resources from multiple providers.
    CodeModeResourceProvider
    llmling_agent.resource_providers.codemode.provider
    Provider that wraps tools into a single Python execution environment.
    Tool
    llmling_agent.tools.base
    Information about a registered tool.

      🛈 DocStrings

      Meta-resource provider that exposes tools through Python execution.

      CodeModeResourceProvider

      Bases: AggregatingResourceProvider

      Provider that wraps tools into a single Python execution environment.

      Source code in src/llmling_agent/resource_providers/codemode/provider.py 
       24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
      class CodeModeResourceProvider(AggregatingResourceProvider):
    """Provider that wraps tools into a single Python execution environment."""

    def __init__(
        self,
        providers: list[ResourceProvider],
        name: str = "meta_tools",
        include_docstrings: bool = True,
        usage_notes: str = USAGE,
    ) -> None:
        """Initialize meta provider.

        Args:
            providers: Providers whose tools to wrap
            name: Provider name
            include_docstrings: Include function docstrings in documentation
            usage_notes: Usage notes for the codemode tool
        """
        super().__init__(providers=providers, name=name)
        self.include_docstrings = include_docstrings
        self._cached_tool: Tool | None = None
        self.usage_notes = usage_notes

    async def get_tools(self) -> list[Tool]:
        """Return single meta-tool for Python execution with available tools."""
        # Always generate fresh toolset to reflect current tools
        toolset_generator = await self._get_code_generator()
        desc = toolset_generator.generate_tool_description()
        desc += self.usage_notes

        if self._cached_tool is None:
            # Create a closure that captures self but isn't a bound method
            async def execute_tool(
                ctx: AgentContext,
                python_code: str,
            ) -> Any:
                """These docstings are overriden by description_override."""
                return await self.execute(ctx, python_code)

            self._cached_tool = Tool.from_callable(execute_tool, description_override=desc)
        else:
            # Update the description on existing cached tool
            self._cached_tool.description = desc

        return [self._cached_tool]

    async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
        """Execute Python code with all wrapped tools available as functions.

        Args:
            python_code: Python code to execute

        Returns:
            Result of the last expression or explicit return value
        """
        toolset_generator = await self._get_code_generator()
        namespace = toolset_generator.generate_execution_namespace()

        # async def report_progress(current: int, total: int, message: str = ""):
        #     """Report progress during code execution."""
        #     await ctx.report_progress(current, total, message)

        # namespace["report_progress"] = NamespaceCallable(report_progress)

        validate_code(python_code)
        try:
            exec(python_code, namespace)
            result = await namespace["main"]()
            # Handle edge cases with coroutines and return values
            if inspect.iscoroutine(result):
                result = await result
            if not result:  # in order to not confuse the model, return a success message.
                return "Code executed successfully"
        except Exception as e:  # noqa: BLE001
            return f"Error executing code: {e!s}"
        else:
            return result

    def invalidate_cache(self) -> None:
        """Invalidate cached tool when providers change."""
        self._cached_tool = None
        # Note: We no longer cache the toolset generator, so no need to clear it

    async def _get_code_generator(self) -> ToolsetCodeGenerator:
        """Get fresh toolset generator with current tools."""
        return tools_to_codegen(
            tools=await super().get_tools(),
            include_docstrings=self.include_docstrings,
        )

      __init__ 

      __init__(
    providers: list[ResourceProvider],
    name: str = "meta_tools",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
) -> None

      Initialize meta provider.

      Parameters:

      Name Type Description Default
      providers list[ResourceProvider]

      Providers whose tools to wrap

      		 required
      name str

      Provider name

      		 'meta_tools'
      include_docstrings bool

      Include function docstrings in documentation

      		 True
      usage_notes str

      Usage notes for the codemode tool

      		 USAGE
      Source code in src/llmling_agent/resource_providers/codemode/provider.py 
      27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
      def __init__(
    self,
    providers: list[ResourceProvider],
    name: str = "meta_tools",
    include_docstrings: bool = True,
    usage_notes: str = USAGE,
) -> None:
    """Initialize meta provider.

    Args:
        providers: Providers whose tools to wrap
        name: Provider name
        include_docstrings: Include function docstrings in documentation
        usage_notes: Usage notes for the codemode tool
    """
    super().__init__(providers=providers, name=name)
    self.include_docstrings = include_docstrings
    self._cached_tool: Tool | None = None
    self.usage_notes = usage_notes

      execute async 

      execute(ctx: AgentContext, python_code: str) -> Any

      Execute Python code with all wrapped tools available as functions.

      Parameters:

      Name Type Description Default
      python_code str

      Python code to execute

      		 required

      Returns:

      Type Description
      Any

      Result of the last expression or explicit return value
      Source code in src/llmling_agent/resource_providers/codemode/provider.py 
       70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
      async def execute(self, ctx: AgentContext, python_code: str) -> Any:  # noqa: D417
    """Execute Python code with all wrapped tools available as functions.

    Args:
        python_code: Python code to execute

    Returns:
        Result of the last expression or explicit return value
    """
    toolset_generator = await self._get_code_generator()
    namespace = toolset_generator.generate_execution_namespace()

    # async def report_progress(current: int, total: int, message: str = ""):
    #     """Report progress during code execution."""
    #     await ctx.report_progress(current, total, message)

    # namespace["report_progress"] = NamespaceCallable(report_progress)

    validate_code(python_code)
    try:
        exec(python_code, namespace)
        result = await namespace["main"]()
        # Handle edge cases with coroutines and return values
        if inspect.iscoroutine(result):
            result = await result
        if not result:  # in order to not confuse the model, return a success message.
            return "Code executed successfully"
    except Exception as e:  # noqa: BLE001
        return f"Error executing code: {e!s}"
    else:
        return result

      get_tools async 

      get_tools() -> list[Tool]

      Return single meta-tool for Python execution with available tools.

      Source code in src/llmling_agent/resource_providers/codemode/provider.py 
      47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
      async def get_tools(self) -> list[Tool]:
    """Return single meta-tool for Python execution with available tools."""
    # Always generate fresh toolset to reflect current tools
    toolset_generator = await self._get_code_generator()
    desc = toolset_generator.generate_tool_description()
    desc += self.usage_notes

    if self._cached_tool is None:
        # Create a closure that captures self but isn't a bound method
        async def execute_tool(
            ctx: AgentContext,
            python_code: str,
        ) -> Any:
            """These docstings are overriden by description_override."""
            return await self.execute(ctx, python_code)

        self._cached_tool = Tool.from_callable(execute_tool, description_override=desc)
    else:
        # Update the description on existing cached tool
        self._cached_tool.description = desc

    return [self._cached_tool]

      invalidate_cache 

      invalidate_cache() -> None

      Invalidate cached tool when providers change.

      Source code in src/llmling_agent/resource_providers/codemode/provider.py 
      102
103
104
      def invalidate_cache(self) -> None:
    """Invalidate cached tool when providers change."""
    self._cached_tool = None