diff --git a/src/strands/tools/mcp/mcp_client.py b/src/strands/tools/mcp/mcp_client.py index c689094..53a93a1 100644 --- a/src/strands/tools/mcp/mcp_client.py +++ b/src/strands/tools/mcp/mcp_client.py @@ -156,6 +156,52 @@ async def _list_tools_async() -> ListToolsResult: mcp_tools = [MCPAgentTool(tool, self) for tool in list_tools_response.tools] self._log_debug_with_thread("successfully adapted %d MCP tools", len(mcp_tools)) return mcp_tools + + def list_prompts_sync(self) -> ListPromptsResult: + """Synchronously retrieves the list of available prompts from the MCP server. + + This method calls the asynchronous list_prompts method on the MCP session + and adapts the returned prompts to the AgentTool interface. + + Returns: + ListPromptsResult: A list of available prompts + """ + self._log_debug_with_thread("listing MCP prompts synchronously") + if not self._is_session_active(): + raise MCPClientInitializationError("the client session is not running") + + async def _list_prompts_async() -> ListPromptsResult: + return await self._background_thread_session.list_prompts() + + list_prompts_response: ListPromptsResult = self._invoke_on_background_thread(_list_prompts_async()) + self._log_debug_with_thread("received %d prompts from MCP server:") + for prompt in list_prompts_response.prompts: + self._log_debug_with_thread(prompt.name) + + return list_prompts_response + + def get_prompt_sync(self, prompt_id: str, args: dict[Any, Any]) -> GetPromptResult: + """Synchronously retrieves a prompt from the MCP server. + + Args: + prompt_id: The ID of the prompt to retrieve + args: Optional arguments to pass to the prompt + + Returns: + GetPromptResult: The prompt response from the MCP server + """ + self._log_debug_with_thread("getting MCP prompt synchronously") + if not self._is_session_active(): + raise MCPClientInitializationError("the client session is not running") + + async def _get_prompt_async(): + return await self._background_thread_session.get_prompt( + prompt_id, arguments=args + ) + get_prompt_response: GetPromptResult = self._invoke_on_background_thread(_get_prompt_async()) + self._log_debug_with_thread("received prompt from MCP server:", get_prompt_response) + + return get_prompt_response def call_tool_sync( self,