Added MCP server prompts functionality

libs/agno/agno/tools/mcp.py

@@ -11,6 +11,9 @@
 from agno.utils.log import log_debug, log_error, log_info, log_warning
 from agno.utils.mcp import get_entrypoint_for_tool
 
+# New addition
+from agno.utils.mcp import get_entrypoint_for_prompt 
+
 try:
     from mcp import ClientSession, StdioServerParameters
     from mcp.client.sse import sse_client
@@ -388,6 +391,62 @@ async def initialize(self) -> None:
             log_debug(f"{self.name} initialized with {len(filtered_tools)} tools")
             self._initialized = True
 
+            # ===== PROMPTS SECTION (new code) =====
+            try:
+                # Get the list of prompts from the MCP server
+                available_prompts = await self.session.list_prompts()
+
+                # Store prompts in a dictionary for easy access
+                self.prompts = {}
+
+                for prompt in available_prompts.prompts:
+                    try:
+                        # Store the prompt metadata
+                        self.prompts[prompt.name] = prompt
+
+                        # Get an entrypoint for the prompt (same pattern as tools!)
+                        entrypoint = get_entrypoint_for_prompt(prompt, self.session)
+
+                        # Build parameters schema from prompt arguments
+                        parameters = {
+                            "type": "object",
+                            "properties": {},
+                            "required": []
+                        }
+
+                        # Add each argument from the prompt to the schema
+                        if prompt.arguments:
+                            for arg in prompt.arguments:
+                                parameters["properties"][arg.name] = {
+                                    "type": "string",
+                                    "description": arg.description if arg.description else f"Argument: {arg.name}"
+                                }
+                                if arg.required:
+                                    parameters["required"].append(arg.name)
+
+                        # Create a Function for the prompt (just like tools!)
+                        f = Function(
+                            name=f"{prompt.name}",
+                            description=prompt.description,
+                            parameters=parameters,
+                            entrypoint=entrypoint,
+                            skip_entrypoint_processing=True,
+                        )                                                                               
+
+                        # Register the Function with the toolkit (THIS IS THE KEY PART!)
+                        self.functions[f.name] = f
+                        log_debug(f"Prompt: {prompt.name} registered with {self.name}")
+
+                    except Exception as e:
+                        log_error(f"Failed to register prompt {prompt.name}: {e}")
+
+                log_debug(f"{self.name} initialized with {len(self.prompts)} prompts")
+
+            except Exception as e:
+                # If the server doesn't support prompts, log a warning but don't fail
+                log_warning(f"Could not fetch prompts from MCP server: {e}")
+                self.prompts = {}
+
         except Exception as e:
             log_error(f"Failed to get MCP tools: {e}")
             raise

libs/agno/agno/utils/mcp.py

@@ -8,6 +8,10 @@
     from mcp import ClientSession
     from mcp.types import CallToolResult, EmbeddedResource, ImageContent, TextContent
     from mcp.types import Tool as MCPTool
+
+    # New additions
+    from mcp.types import Prompt as MCPPrompt
+    from mcp.types import GetPromptResult  
 except (ImportError, ModuleNotFoundError):
     raise ImportError("`mcp` not installed. Please install using `pip install mcp`")
 
@@ -122,3 +126,48 @@ async def call_tool(agent: Agent, tool_name: str, **kwargs) -> ToolResult:
             return ToolResult(content=f"Error: {e}")
 
     return partial(call_tool, tool_name=tool.name)
+
+# ADD THIS NEW FUNCTION
+def get_entrypoint_for_prompt(prompt: MCPPrompt, session: ClientSession):
+    """
+    Return an entrypoint for an MCP prompt.
+
+    Args:
+        prompt: The MCP prompt to create an entrypoint for
+        session: The session to use
+
+    Returns:
+        Callable: The entrypoint function for the prompt
+    """
+    from agno.agent import Agent
+
+    async def get_prompt(agent: Agent, prompt_name: str, **kwargs) -> str:
+        try:
+            log_debug(f"Retrieving MCP Prompt '{prompt_name}' with args: {kwargs}")
+
+            # Convert all arguments to strings (get_prompt expects dict[str, str])
+            arguments = {k: str(v) for k, v in kwargs.items()} if kwargs else None
+
+            # Call the MCP get_prompt method
+            result: GetPromptResult = await session.get_prompt(prompt_name, arguments)  # type: ignore
+
+            # Extract text content from all messages
+            messages = []
+            for message in result.messages:
+                # Handle different content types
+                if hasattr(message.content, 'text'):
+                    messages.append(message.content.text)
+                elif isinstance(message.content, TextContent):
+                    messages.append(message.content.text)
+                else:
+                    # Fallback for other content types
+                    messages.append(str(message.content))
+
+            # Join all messages with double newlines
+            return "\n\n".join(messages) if messages else ""
+
+        except Exception as e:
+            log_exception(f"Failed to retrieve MCP prompt '{prompt_name}': {e}")
+            return f"Error retrieving prompt: {e}"
+
+    return partial(get_prompt, prompt_name=prompt.name)