Support mcp prompts (#1010)

### Summary

Add MCP prompt support to enable user-controlled agent instruction
generation. This enhancement allows MCP servers to provide prompts that
can be used to dynamically generate agent instructions, enabling more
flexible and user-controlled agent behavior.

**Key changes:**
- Added abstract `list_prompts()` and `get_prompt()` methods to base
`MCPServer` class
- Implemented prompt support in `MCPServerStdio` and
`MCPServerStreamableHttp` classes
- Added comprehensive test suite for prompt functionality
- Created example MCP prompt server with working demonstration
- Updated documentation with prompt usage examples

**Note:** This PR implements MCP prompt support only. Resource support
is not included in this implementation.

### Test plan

- **Unit tests**: Added 11 comprehensive tests in
`tests/mcp/test_prompt_server.py` covering:
  - Prompt listing and retrieval
  - Argument formatting and validation
  - Agent integration with prompt-generated instructions
  - Streaming and non-streaming scenarios
  - Error handling for missing prompts
- **Integration tests**: Updated existing MCP test suite to handle new
abstract methods
- **Example verification**: Created working example with MCP prompt
server and client
- **All tests pass**: 450/450 tests passing after adding optional
dependencies

### Issue number

Partially addresses #544 (prompts only, resources not implemented)

### Checks

- [x] I've added new tests (if relevant)
- [x] I've added/updated the relevant documentation
- [x] I've run `make lint` and `make format`
- [x] I've made sure tests pass

---------

Co-authored-by: thein <[email protected]>

Author: thoo

docs/mcp.md

@@ -4,7 +4,7 @@ The [Model context protocol](https://modelcontextprotocol.io/introduction) (aka
 
 > MCP is an open protocol that standardizes how applications provide context to LLMs. Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools.
 
-The Agents SDK has support for MCP. This enables you to use a wide range of MCP servers to provide tools to your Agents.
+The Agents SDK has support for MCP. This enables you to use a wide range of MCP servers to provide tools and prompts to your Agents.
 
 ## MCP servers
 
@@ -135,6 +135,38 @@ The `ToolFilterContext` provides access to:
 - `agent`: The agent requesting the tools 
 - `server_name`: The name of the MCP server
 
+## Prompts
+
+MCP servers can also provide prompts that can be used to dynamically generate agent instructions. This allows you to create reusable instruction templates that can be customized with parameters.
+
+### Using prompts
+
+MCP servers that support prompts provide two key methods:
+
+- `list_prompts()`: Lists all available prompts on the server
+- `get_prompt(name, arguments)`: Gets a specific prompt with optional parameters
+
+```python
+# List available prompts
+prompts_result = await server.list_prompts()
+for prompt in prompts_result.prompts:
+    print(f"Prompt: {prompt.name} - {prompt.description}")
+
+# Get a specific prompt with parameters
+prompt_result = await server.get_prompt(
+    "generate_code_review_instructions",
+    {"focus": "security vulnerabilities", "language": "python"}
+)
+instructions = prompt_result.messages[0].content.text
+
+# Use the prompt-generated instructions with an Agent
+agent = Agent(
+    name="Code Reviewer",
+    instructions=instructions,  # Instructions from MCP prompt
+    mcp_servers=[server]
+)
+```
+
 ## Caching
 
 Every time an Agent runs, it calls `list_tools()` on the MCP server. This can be a latency hit, especially if the server is a remote server. To automatically cache the list of tools, you can pass `cache_tools_list=True` to [`MCPServerStdio`][agents.mcp.server.MCPServerStdio], [`MCPServerSse`][agents.mcp.server.MCPServerSse], and [`MCPServerStreamableHttp`][agents.mcp.server.MCPServerStreamableHttp]. You should only do this if you're certain the tool list will not change.

examples/mcp/prompt_server/README.md

@@ -0,0 +1,29 @@
+# MCP Prompt Server Example
+
+This example uses a local MCP prompt server in [server.py](server.py).
+
+Run the example via:
+
+```
+uv run python examples/mcp/prompt_server/main.py
+```
+
+## Details
+
+The example uses the `MCPServerStreamableHttp` class from `agents.mcp`. The server runs in a sub-process at `http://localhost:8000/mcp` and provides user-controlled prompts that generate agent instructions.
+
+The server exposes prompts like `generate_code_review_instructions` that take parameters such as focus area and programming language. The agent calls these prompts to dynamically generate its system instructions based on user-provided parameters.
+
+## Workflow
+
+The example demonstrates two key functions:
+
+1. **`show_available_prompts`** - Lists all available prompts on the MCP server, showing users what prompts they can select from. This demonstrates the discovery aspect of MCP prompts.
+
+2. **`demo_code_review`** - Shows the complete user-controlled prompt workflow:
+   - Calls `generate_code_review_instructions` with specific parameters (focus: "security vulnerabilities", language: "python")
+   - Uses the generated instructions to create an Agent with specialized code review capabilities
+   - Runs the agent against vulnerable sample code (command injection via `os.system`)
+   - The agent analyzes the code and provides security-focused feedback using available tools
+
+This pattern allows users to dynamically configure agent behavior through MCP prompts rather than hardcoded instructions. 

examples/mcp/prompt_server/main.py

@@ -0,0 +1,110 @@
+import asyncio
+import os
+import shutil
+import subprocess
+import time
+from typing import Any
+
+from agents import Agent, Runner, gen_trace_id, trace
+from agents.mcp import MCPServer, MCPServerStreamableHttp
+from agents.model_settings import ModelSettings
+
+
+async def get_instructions_from_prompt(mcp_server: MCPServer, prompt_name: str, **kwargs) -> str:
+    """Get agent instructions by calling MCP prompt endpoint (user-controlled)"""
+    print(f"Getting instructions from prompt: {prompt_name}")
+
+    try:
+        prompt_result = await mcp_server.get_prompt(prompt_name, kwargs)
+        content = prompt_result.messages[0].content
+        if hasattr(content, 'text'):
+            instructions = content.text
+        else:
+            instructions = str(content)
+        print("Generated instructions")
+        return instructions
+    except Exception as e:
+        print(f"Failed to get instructions: {e}")
+        return f"You are a helpful assistant. Error: {e}"
+
+
+async def demo_code_review(mcp_server: MCPServer):
+    """Demo: Code review with user-selected prompt"""
+    print("=== CODE REVIEW DEMO ===")
+
+    # User explicitly selects prompt and parameters
+    instructions = await get_instructions_from_prompt(
+        mcp_server,
+        "generate_code_review_instructions",
+        focus="security vulnerabilities",
+        language="python",
+    )
+
+    agent = Agent(
+        name="Code Reviewer Agent",
+        instructions=instructions,  # Instructions from MCP prompt
+        model_settings=ModelSettings(tool_choice="auto"),
+    )
+
+    message = """Please review this code:
+
+def process_user_input(user_input):
+    command = f"echo {user_input}"
+    os.system(command)
+    return "Command executed"
+
+"""
+
+    print(f"Running: {message[:60]}...")
+    result = await Runner.run(starting_agent=agent, input=message)
+    print(result.final_output)
+    print("\n" + "=" * 50 + "\n")
+
+
+async def show_available_prompts(mcp_server: MCPServer):
+    """Show available prompts for user selection"""
+    print("=== AVAILABLE PROMPTS ===")
+
+    prompts_result = await mcp_server.list_prompts()
+    print("User can select from these prompts:")
+    for i, prompt in enumerate(prompts_result.prompts, 1):
+        print(f"  {i}. {prompt.name} - {prompt.description}")
+    print()
+
+
+async def main():
+    async with MCPServerStreamableHttp(
+        name="Simple Prompt Server",
+        params={"url": "http://localhost:8000/mcp"},
+    ) as server:
+        trace_id = gen_trace_id()
+        with trace(workflow_name="Simple Prompt Demo", trace_id=trace_id):
+            print(f"Trace: https://platform.openai.com/traces/trace?trace_id={trace_id}\n")
+
+            await show_available_prompts(server)
+            await demo_code_review(server)
+
+
+if __name__ == "__main__":
+    if not shutil.which("uv"):
+        raise RuntimeError("uv is not installed")
+
+    process: subprocess.Popen[Any] | None = None
+    try:
+        this_dir = os.path.dirname(os.path.abspath(__file__))
+        server_file = os.path.join(this_dir, "server.py")
+
+        print("Starting Simple Prompt Server...")
+        process = subprocess.Popen(["uv", "run", server_file])
+        time.sleep(3)
+        print("Server started\n")
+    except Exception as e:
+        print(f"Error starting server: {e}")
+        exit(1)
+
+    try:
+        asyncio.run(main())
+    finally:
+        if process:
+            process.terminate()
+            print("Server terminated.")

examples/mcp/prompt_server/server.py

@@ -0,0 +1,37 @@
+from mcp.server.fastmcp import FastMCP
+
+# Create server
+mcp = FastMCP("Prompt Server")
+
+
+# Instruction-generating prompts (user-controlled)
+@mcp.prompt()
+def generate_code_review_instructions(
+    focus: str = "general code quality", language: str = "python"
+) -> str:
+    """Generate agent instructions for code review tasks"""
+    print(f"[debug-server] generate_code_review_instructions({focus}, {language})")
+
+    return f"""You are a senior {language} code review specialist. Your role is to provide comprehensive code analysis with focus on {focus}.
+
+INSTRUCTIONS:
+- Analyze code for quality, security, performance, and best practices
+- Provide specific, actionable feedback with examples
+- Identify potential bugs, vulnerabilities, and optimization opportunities
+- Suggest improvements with code examples when applicable
+- Be constructive and educational in your feedback
+- Focus particularly on {focus} aspects
+
+RESPONSE FORMAT:
+1. Overall Assessment
+2. Specific Issues Found
+3. Security Considerations
+4. Performance Notes
+5. Recommended Improvements
+6. Best Practices Suggestions
+
+Use the available tools to check current time if you need timestamps for your analysis."""
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")

src/agents/mcp/server.py

@@ -13,7 +13,7 @@
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import GetSessionIdCallback, streamablehttp_client
 from mcp.shared.message import SessionMessage
-from mcp.types import CallToolResult, InitializeResult
+from mcp.types import CallToolResult, GetPromptResult, InitializeResult, ListPromptsResult
 from typing_extensions import NotRequired, TypedDict
 
 from ..exceptions import UserError
@@ -63,6 +63,20 @@ async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> C
         """Invoke a tool on the server."""
         pass
 
+    @abc.abstractmethod
+    async def list_prompts(
+        self,
+    ) -> ListPromptsResult:
+        """List the prompts available on the server."""
+        pass
+
+    @abc.abstractmethod
+    async def get_prompt(
+        self, name: str, arguments: dict[str, Any] | None = None
+    ) -> GetPromptResult:
+        """Get a specific prompt from the server."""
+        pass
+
 
 class _MCPServerWithClientSession(MCPServer, abc.ABC):
     """Base class for MCP servers that use a `ClientSession` to communicate with the server."""
@@ -118,9 +132,7 @@ async def _apply_tool_filter(
             return await self._apply_dynamic_tool_filter(tools, run_context, agent)
 
     def _apply_static_tool_filter(
-        self,
-        tools: list[MCPTool],
-        static_filter: ToolFilterStatic
+        self, tools: list[MCPTool], static_filter: ToolFilterStatic
     ) -> list[MCPTool]:
         """Apply static tool filtering based on allowlist and blocklist."""
         filtered_tools = tools
@@ -261,6 +273,24 @@ async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> C
 
         return await self.session.call_tool(tool_name, arguments)
 
+    async def list_prompts(
+        self,
+    ) -> ListPromptsResult:
+        """List the prompts available on the server."""
+        if not self.session:
+            raise UserError("Server not initialized. Make sure you call `connect()` first.")
+
+        return await self.session.list_prompts()
+
+    async def get_prompt(
+        self, name: str, arguments: dict[str, Any] | None = None
+    ) -> GetPromptResult:
+        """Get a specific prompt from the server."""
+        if not self.session:
+            raise UserError("Server not initialized. Make sure you call `connect()` first.")
+
+        return await self.session.get_prompt(name, arguments)
+
     async def cleanup(self):
         """Cleanup the server."""
         async with self._cleanup_lock:

tests/mcp/helpers.py

@@ -4,7 +4,7 @@
 from typing import Any
 
 from mcp import Tool as MCPTool
-from mcp.types import CallToolResult, TextContent
+from mcp.types import CallToolResult, GetPromptResult, ListPromptsResult, PromptMessage, TextContent
 
 from agents.mcp import MCPServer
 from agents.mcp.server import _MCPServerWithClientSession
@@ -94,6 +94,18 @@ async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> C
             content=[TextContent(text=self.tool_results[-1], type="text")],
         )
 
+    async def list_prompts(self, run_context=None, agent=None) -> ListPromptsResult:
+        """Return empty list of prompts for fake server"""
+        return ListPromptsResult(prompts=[])
+
+    async def get_prompt(
+        self, name: str, arguments: dict[str, Any] | None = None
+    ) -> GetPromptResult:
+        """Return a simple prompt result for fake server"""
+        content = f"Fake prompt content for {name}"
+        message = PromptMessage(role="user", content=TextContent(type="text", text=content))
+        return GetPromptResult(description=f"Fake prompt: {name}", messages=[message])
+
     @property
     def name(self) -> str:
         return self._server_name