Skip to content

feat: Add list_prompts, get_prompt methods #160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: Add list_prompts, get_prompt methods #160

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions src/strands/tools/mcp/mcp_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Copy link
Member

@dbschmigelski dbschmigelski Jun 2, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I want to thank you for proposing these changes. We appreciate the contributions and definitely are eager to expose MCP prompt features in Strands.

I think there are a few things missing here relating to the complete flow of mcp.

The Agent interacts with MCP servers through tools which are exposed via the AgentTool interface. In our case list_tools_sync is used to retrieve a list of AgentTools. These are then provided to the agent like

with streamable_http_mcp_client:
    tools = streamable_http_mcp_client.list_tools_sync()
    agent = Agent(tools=tools)

So, I think what is missing from this PR is the "how" of how the prompts will be exposed to the Agent as right now the agent lacks a mechanism to list and get the prompts. To address this, we should create an MCPPromptTool similar to the MCPAgentTool. The naming may be confusing here, but in essence, anything exposed to the Agent class must be via a tool. In this case, a "tool" available to the Agent will be the ability to fetch a prompt. The response of a get prompt is itself a ToolResult. There is an argument for making prompts first class citizens in the Agent class. But to expose this functionality now I believe this will be the best path forward.

In addition to that, we should add unit tests in the tests/tools/mcp directory and modify the integration tests in tests-integ/test_mcp_client.py.

"""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:
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems like these MCP response types are missing imports

"""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,
Expand Down