diff --git a/libs/agno/agno/tools/mcp.py b/libs/agno/agno/tools/mcp.py index 13612cc6d31..26732d0d111 100644 --- a/libs/agno/agno/tools/mcp.py +++ b/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 diff --git a/libs/agno/agno/utils/mcp.py b/libs/agno/agno/utils/mcp.py index fd5dd730a13..beaf4c90652 100644 --- a/libs/agno/agno/utils/mcp.py +++ b/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) \ No newline at end of file