Dynamic toolsets based on run context

[strawgate]

Description

When providing a toolset which is dynamic like a wrapper toolset or an MCP server which needs a run-context-specific configuration, it would be really great if the toolset could be created and made available to the agent during the run similar to how the system prompt and instructions are.

I.e. a function that takes runcontext and can add toolsets to the agent when it runs

References

No response

[strawgate]

For example, I have a Filesystem MCP Server which has a --root-dir argument which I want to set to the path to the code repository, but I dont know the path to the code repository until the Agent is invoked.

[DouweM]

@strawgate Thanks for pushing toolsets to the limits and filing these issues!

Have you considered defining a AbstractToolset subclass that will dynamically build an MCPServer based on RunContext in get_tools()? To make call_tool() work, you'd want to have get_tools() return new wrapper ToolsetTool objects for each tool that hold a reference to the new toolset that call_tool() should be called on, similar to how CombinedToolset does it, as the dynamic toolset can be called with different run contexts.

The only downside is that the MCP server connection wouldn't be initialized when the agent or wrapper toolset are entered in around the run, as there's no run context yet at that point, but I don't think there's much we can do about that...

That wrapper toolset would be nice to include with Pydantic AI as DynamicToolset, which could take a function that generates a toolset (based on ctx), or could be subclassed to define it as a method.

[strawgate]

I am working on a set of Toolsets that make it easier to bridge FastMCP tools into Pydantic AI and that is exactly what I'm doing:

    async def _setup_fastmcp_server(self, ctx: RunContext[AgentDepsT]) -> None:
        customized_mcp_config: MCPConfig = await self._prepare_mcp_config_fn(ctx, self.mcp_config)

        fastmcp_server = composite_server_from_mcp_config(config=customized_mcp_config, name_as_prefix=False)

        self._running_fastmcp_server.set(fastmcp_server)

    async def get_tools(self, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
        if not self._running_fastmcp_server.get():
            await self._setup_fastmcp_server(ctx=ctx)

        return await super().get_tools(ctx=ctx)

I was using a ContextVar to store the server.

Almost like a wrapper toolset which stores the wrapped toolset in a context var and provides a "setup_toolset()` abstract method?

I can noodle on it a little more and review the combined toolset and maybe make a draft pr

[strawgate]

@DouweM I think we need to store the toolset in a contextvar to avoid making a new server every time we call get_tools and also allow multiple run contexts

I think it will require a contextvar that gets initially populated when the context gets entered and then on the first call to get_tools the contextvar gets populated with the toolset from the dynamic function.

Unfortunately the agent is not entering its own context so this requires an explicit async with on the agent

I kinda have this working by making an empty combined toolset in a context var and then running the dynamic function when the combined toolset is empty -- will post a bad example later

[DouweM]

@strawgate Will that work correctly if the agent (and thus the toolset) are entered at a higher scope that covers multiple agent runs? Maybe the dynamic toolset should keep track of which toolset to use based on the RunContext value that was used to generate it, like the repo path in your case... Either way I'd love to see some code to make the discussion more concrete!