Skip to content

Support mcp prompts #1010

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.

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 6 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
34 changes: 33 additions & 1 deletion docs/mcp.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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.
Expand Down
29 changes: 29 additions & 0 deletions examples/mcp/prompt_server/README.md
Original file line number Diff line number Diff line change
@@ -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.
110 changes: 110 additions & 0 deletions examples/mcp/prompt_server/main.py
Original file line number Diff line number Diff line change
@@ -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.")
37 changes: 37 additions & 0 deletions examples/mcp/prompt_server/server.py
Original file line number Diff line number Diff line change
@@ -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")
38 changes: 34 additions & 4 deletions src/agents/mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
14 changes: 13 additions & 1 deletion tests/mcp/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Loading