feat(mcp): add experimental agent managed connection via ToolProvider (#895)

Author: dbschmigelski

.codecov.yml

@@ -0,0 +1,11 @@
+coverage:
+  status:
+    project:
+      default:
+        target: 90%    # overall coverage threshold
+    patch:
+      default:
+        target: 90%    # patch coverage threshold
+        base: auto
+        # Only post patch coverage on decreases
+        only_pulls: true

src/strands/_async.py

@@ -0,0 +1,31 @@
+"""Private async execution utilities."""
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from typing import Awaitable, Callable, TypeVar
+
+T = TypeVar("T")
+
+
+def run_async(async_func: Callable[[], Awaitable[T]]) -> T:
+    """Run an async function in a separate thread to avoid event loop conflicts.
+
+    This utility handles the common pattern of running async code from sync contexts
+    by using ThreadPoolExecutor to isolate the async execution.
+
+    Args:
+        async_func: A callable that returns an awaitable
+
+    Returns:
+        The result of the async function
+    """
+
+    async def execute_async() -> T:
+        return await async_func()
+
+    def execute() -> T:
+        return asyncio.run(execute_async())
+
+    with ThreadPoolExecutor() as executor:
+        future = executor.submit(execute)
+        return future.result()

src/strands/agent/agent.py

@@ -9,13 +9,12 @@
 2. Method-style for direct tool access: `agent.tool.tool_name(param1="value")`
 """
 
-import asyncio
 import json
 import logging
 import random
 import warnings
-from concurrent.futures import ThreadPoolExecutor
 from typing import (
+    TYPE_CHECKING,
     Any,
     AsyncGenerator,
     AsyncIterator,
@@ -32,7 +31,11 @@
 from pydantic import BaseModel
 
 from .. import _identifier
+from .._async import run_async
 from ..event_loop.event_loop import event_loop_cycle
+
+if TYPE_CHECKING:
+    from ..experimental.tools import ToolProvider
 from ..handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
 from ..hooks import (
     AfterInvocationEvent,
@@ -167,12 +170,7 @@ async def acall() -> ToolResult:
 
                     return tool_results[0]
 
-                def tcall() -> ToolResult:
-                    return asyncio.run(acall())
-
-                with ThreadPoolExecutor() as executor:
-                    future = executor.submit(tcall)
-                    tool_result = future.result()
+                tool_result = run_async(acall)
 
                 if record_direct_tool_call is not None:
                     should_record_direct_tool_call = record_direct_tool_call
@@ -215,7 +213,7 @@ def __init__(
         self,
         model: Union[Model, str, None] = None,
         messages: Optional[Messages] = None,
-        tools: Optional[list[Union[str, dict[str, str], Any]]] = None,
+        tools: Optional[list[Union[str, dict[str, str], "ToolProvider", Any]]] = None,
         system_prompt: Optional[str] = None,
         structured_output_model: Optional[Type[BaseModel]] = None,
         callback_handler: Optional[
@@ -248,6 +246,7 @@ def __init__(
                 - File paths (e.g., "/path/to/tool.py")
                 - Imported Python modules (e.g., from strands_tools import current_time)
                 - Dictionaries with name/path keys (e.g., {"name": "tool_name", "path": "/path/to/tool.py"})
+                - ToolProvider instances for managed tool collections
                 - Functions decorated with `@strands.tool` decorator.
 
                 If provided, only these tools will be available. If None, all tools will be available.
@@ -423,17 +422,11 @@ def __call__(
                 - state: The final state of the event loop
                 - structured_output: Parsed structured output when structured_output_model was specified
         """
-
-        def execute() -> AgentResult:
-            return asyncio.run(
-                self.invoke_async(
-                    prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
-                )
+        return run_async(
+            lambda: self.invoke_async(
+                prompt, invocation_state=invocation_state, structured_output_model=structured_output_model, **kwargs
             )
-
-        with ThreadPoolExecutor() as executor:
-            future = executor.submit(execute)
-            return future.result()
+        )
 
     async def invoke_async(
         self,
@@ -506,12 +499,7 @@ def structured_output(self, output_model: Type[T], prompt: AgentInput = None) ->
             stacklevel=2,
         )
 
-        def execute() -> T:
-            return asyncio.run(self.structured_output_async(output_model, prompt))
-
-        with ThreadPoolExecutor() as executor:
-            future = executor.submit(execute)
-            return future.result()
+        return run_async(lambda: self.structured_output_async(output_model, prompt))
 
     async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
         """This method allows you to get structured output from the agent.
@@ -529,6 +517,7 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
 
         Raises:
             ValueError: If no conversation history or prompt is provided.
+        -
         """
         if self._interrupt_state.activated:
             raise RuntimeError("cannot call structured output during interrupt")
@@ -583,6 +572,25 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
             finally:
                 self.hooks.invoke_callbacks(AfterInvocationEvent(agent=self))
 
+    def cleanup(self) -> None:
+        """Clean up resources used by the agent.
+
+        This method cleans up all tool providers that require explicit cleanup,
+        such as MCP clients. It should be called when the agent is no longer needed
+        to ensure proper resource cleanup.
+
+        Note: This method uses a "belt and braces" approach with automatic cleanup
+        through finalizers as a fallback, but explicit cleanup is recommended.
+        """
+        self.tool_registry.cleanup()
+
+    def __del__(self) -> None:
+        """Clean up resources when agent is garbage collected."""
+        # __del__ is called even when an exception is thrown in the constructor,
+        # so there is no guarantee tool_registry was set..
+        if hasattr(self, "tool_registry"):
+            self.tool_registry.cleanup()
+
     async def stream_async(
         self,
         prompt: AgentInput = None,

src/strands/experimental/__init__.py

@@ -3,6 +3,7 @@
 This module implements experimental features that are subject to change in future revisions without notice.
 """
 
+from . import tools
 from .agent_config import config_to_agent
 
-__all__ = ["config_to_agent"]
+__all__ = ["config_to_agent", "tools"]

src/strands/experimental/agent_config.py

@@ -18,8 +18,6 @@
 import jsonschema
 from jsonschema import ValidationError
 
-from ..agent import Agent
-
 # JSON Schema for agent configuration
 AGENT_CONFIG_SCHEMA = {
     "$schema": "http://json-schema.org/draft-07/schema#",
@@ -53,7 +51,7 @@
 _VALIDATOR = jsonschema.Draft7Validator(AGENT_CONFIG_SCHEMA)
 
 
-def config_to_agent(config: str | dict[str, Any], **kwargs: dict[str, Any]) -> Agent:
+def config_to_agent(config: str | dict[str, Any], **kwargs: dict[str, Any]) -> Any:
     """Create an Agent from a configuration file or dictionary.
 
     This function supports tools that can be loaded declaratively (file paths, module names,
@@ -134,5 +132,8 @@ def config_to_agent(config: str | dict[str, Any], **kwargs: dict[str, Any]) -> A
     # Override with any additional kwargs provided
     agent_kwargs.update(kwargs)
 
+    # Import Agent at runtime to avoid circular imports
+    from ..agent import Agent
+
     # Create and return Agent
     return Agent(**agent_kwargs)

src/strands/experimental/tools/__init__.py

@@ -0,0 +1,5 @@
+"""Experimental tools package."""
+
+from .tool_provider import ToolProvider
+
+__all__ = ["ToolProvider"]

src/strands/experimental/tools/tool_provider.py

@@ -0,0 +1,52 @@
+"""Tool provider interface."""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Sequence
+
+if TYPE_CHECKING:
+    from ...types.tools import AgentTool
+
+
+class ToolProvider(ABC):
+    """Interface for providing tools with lifecycle management.
+
+    Provides a way to load a collection of tools and clean them up
+    when done, with lifecycle managed by the agent.
+    """
+
+    @abstractmethod
+    async def load_tools(self, **kwargs: Any) -> Sequence["AgentTool"]:
+        """Load and return the tools in this provider.
+
+        Args:
+            **kwargs: Additional arguments for future compatibility.
+
+        Returns:
+            List of tools that are ready to use.
+        """
+        pass
+
+    @abstractmethod
+    def add_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
+        """Add a consumer to this tool provider.
+
+        Args:
+            consumer_id: Unique identifier for the consumer.
+            **kwargs: Additional arguments for future compatibility.
+        """
+        pass
+
+    @abstractmethod
+    def remove_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
+        """Remove a consumer from this tool provider.
+
+        This method must be idempotent - calling it multiple times with the same ID
+        should have no additional effect after the first call.
+
+        Provider may clean up resources when no consumers remain.
+
+        Args:
+            consumer_id: Unique identifier for the consumer.
+            **kwargs: Additional arguments for future compatibility.
+        """
+        pass

src/strands/multiagent/base.py

@@ -3,15 +3,14 @@
 Provides minimal foundation for multi-agent patterns (Swarm, Graph).
 """
 
-import asyncio
 import logging
 import warnings
 from abc import ABC, abstractmethod
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Union
 
+from .._async import run_async
 from ..agent import AgentResult
 from ..types.content import ContentBlock
 from ..types.event_loop import Metrics, Usage
@@ -199,12 +198,7 @@ def __call__(
             invocation_state.update(kwargs)
             warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
 
-        def execute() -> MultiAgentResult:
-            return asyncio.run(self.invoke_async(task, invocation_state))
-
-        with ThreadPoolExecutor() as executor:
-            future = executor.submit(execute)
-            return future.result()
+        return run_async(lambda: self.invoke_async(task, invocation_state))
 
     def serialize_state(self) -> dict[str, Any]:
         """Return a JSON-serializable snapshot of the orchestrator state."""

src/strands/multiagent/graph.py

@@ -18,12 +18,12 @@
 import copy
 import logging
 import time
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from typing import Any, Callable, Optional, Tuple
 
 from opentelemetry import trace as trace_api
 
+from .._async import run_async
 from ..agent import Agent
 from ..agent.state import AgentState
 from ..telemetry import get_tracer
@@ -399,12 +399,7 @@ def __call__(
         if invocation_state is None:
             invocation_state = {}
 
-        def execute() -> GraphResult:
-            return asyncio.run(self.invoke_async(task, invocation_state))
-
-        with ThreadPoolExecutor() as executor:
-            future = executor.submit(execute)
-            return future.result()
+        return run_async(lambda: self.invoke_async(task, invocation_state))
 
     async def invoke_async(
         self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any

src/strands/multiagent/swarm.py

@@ -17,13 +17,14 @@
 import json
 import logging
 import time
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from typing import Any, Callable, Tuple
 
 from opentelemetry import trace as trace_api
 
-from ..agent import Agent, AgentResult
+from .._async import run_async
+from ..agent import Agent
+from ..agent.agent_result import AgentResult
 from ..agent.state import AgentState
 from ..telemetry import get_tracer
 from ..tools.decorator import tool
@@ -254,12 +255,7 @@ def __call__(
         if invocation_state is None:
             invocation_state = {}
 
-        def execute() -> SwarmResult:
-            return asyncio.run(self.invoke_async(task, invocation_state))
-
-        with ThreadPoolExecutor() as executor:
-            future = executor.submit(execute)
-            return future.result()
+        return run_async(lambda: self.invoke_async(task, invocation_state))
 
     async def invoke_async(
         self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any