Skip to content

Index

Prompt

Bases: BaseModel

A prompt template that can be rendered with parameters.

Source code in src/mcp/server/mcpserver/prompts/base.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
class Prompt(BaseModel):
    """A prompt template that can be rendered with parameters."""

    name: str = Field(description="Name of the prompt")
    title: str | None = Field(None, description="Human-readable title of the prompt")
    description: str | None = Field(None, description="Description of what the prompt does")
    arguments: list[PromptArgument] | None = Field(None, description="Arguments that can be passed to the prompt")
    fn: Callable[..., PromptResult | Awaitable[PromptResult]] = Field(exclude=True)
    icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this prompt")
    context_kwarg: str | None = Field(None, description="Name of the kwarg that should receive context", exclude=True)

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., PromptResult | Awaitable[PromptResult]],
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        icons: list[Icon] | None = None,
        context_kwarg: str | None = None,
    ) -> Prompt:
        """Create a Prompt from a function.

        The function can return:
        - A string (converted to a message)
        - A Message object
        - A dict (converted to a message)
        - A sequence of any of the above
        """
        func_name = name or fn.__name__

        if func_name == "<lambda>":  # pragma: no cover
            raise ValueError("You must provide a name for lambda functions")

        # Find context parameter if it exists
        if context_kwarg is None:  # pragma: no branch
            context_kwarg = find_context_parameter(fn)

        # Get schema from func_metadata, excluding context parameter
        func_arg_metadata = func_metadata(
            fn,
            skip_names=[context_kwarg] if context_kwarg is not None else [],
        )
        parameters = func_arg_metadata.arg_model.model_json_schema()

        # Convert parameters to PromptArguments
        arguments: list[PromptArgument] = []
        if "properties" in parameters:  # pragma: no branch
            for param_name, param in parameters["properties"].items():
                required = param_name in parameters.get("required", [])
                arguments.append(
                    PromptArgument(
                        name=param_name,
                        description=param.get("description"),
                        required=required,
                    )
                )

        # ensure the arguments are properly cast
        fn = validate_call(fn)

        return cls(
            name=func_name,
            title=title,
            description=description or fn.__doc__ or "",
            arguments=arguments,
            fn=fn,
            icons=icons,
            context_kwarg=context_kwarg,
        )

    async def render(
        self,
        arguments: dict[str, Any] | None,
        context: Context[LifespanContextT, RequestT],
    ) -> list[Message]:
        """Render the prompt with arguments.

        Raises:
            ValueError: If required arguments are missing, or if rendering fails.
        """
        # Validate required arguments
        if self.arguments:
            required = {arg.name for arg in self.arguments if arg.required}
            provided = set(arguments or {})
            missing = required - provided
            if missing:
                raise ValueError(f"Missing required arguments: {missing}")

        try:
            # Add context to arguments if needed
            call_args = inject_context(self.fn, arguments or {}, context, self.context_kwarg)

            fn = self.fn
            if is_async_callable(fn):
                result = await fn(**call_args)
            else:
                result = await anyio.to_thread.run_sync(functools.partial(self.fn, **call_args))

            # Validate messages
            if not isinstance(result, list | tuple):
                result = [result]

            # Convert result to messages
            messages: list[Message] = []
            for msg in result:  # type: ignore[reportUnknownVariableType]
                try:
                    if isinstance(msg, Message):
                        messages.append(msg)
                    elif isinstance(msg, dict):
                        messages.append(message_validator.validate_python(msg))
                    elif isinstance(msg, str):
                        content = TextContent(type="text", text=msg)
                        messages.append(UserMessage(content=content))
                    else:  # pragma: no cover
                        content = pydantic_core.to_json(msg, fallback=str, indent=2).decode()
                        messages.append(Message(role="user", content=content))
                except Exception:  # pragma: no cover
                    raise ValueError(f"Could not convert prompt result to message: {msg}")

            return messages
        except Exception as e:  # pragma: no cover
            raise ValueError(f"Error rendering prompt {self.name}: {e}")

from_function classmethod 

from_function(
    fn: Callable[
        ..., PromptResult | Awaitable[PromptResult]
    ],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    icons: list[Icon] | None = None,
    context_kwarg: str | None = None,
) -> Prompt

Create a Prompt from a function.

The function can return: - A string (converted to a message) - A Message object - A dict (converted to a message) - A sequence of any of the above

Source code in src/mcp/server/mcpserver/prompts/base.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@classmethod
def from_function(
    cls,
    fn: Callable[..., PromptResult | Awaitable[PromptResult]],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    icons: list[Icon] | None = None,
    context_kwarg: str | None = None,
) -> Prompt:
    """Create a Prompt from a function.

    The function can return:
    - A string (converted to a message)
    - A Message object
    - A dict (converted to a message)
    - A sequence of any of the above
    """
    func_name = name or fn.__name__

    if func_name == "<lambda>":  # pragma: no cover
        raise ValueError("You must provide a name for lambda functions")

    # Find context parameter if it exists
    if context_kwarg is None:  # pragma: no branch
        context_kwarg = find_context_parameter(fn)

    # Get schema from func_metadata, excluding context parameter
    func_arg_metadata = func_metadata(
        fn,
        skip_names=[context_kwarg] if context_kwarg is not None else [],
    )
    parameters = func_arg_metadata.arg_model.model_json_schema()

    # Convert parameters to PromptArguments
    arguments: list[PromptArgument] = []
    if "properties" in parameters:  # pragma: no branch
        for param_name, param in parameters["properties"].items():
            required = param_name in parameters.get("required", [])
            arguments.append(
                PromptArgument(
                    name=param_name,
                    description=param.get("description"),
                    required=required,
                )
            )

    # ensure the arguments are properly cast
    fn = validate_call(fn)

    return cls(
        name=func_name,
        title=title,
        description=description or fn.__doc__ or "",
        arguments=arguments,
        fn=fn,
        icons=icons,
        context_kwarg=context_kwarg,
    )

render async 

render(
    arguments: dict[str, Any] | None,
    context: Context[LifespanContextT, RequestT],
) -> list[Message]

Render the prompt with arguments.

Raises:

Type Description
ValueError

If required arguments are missing, or if rendering fails.
Source code in src/mcp/server/mcpserver/prompts/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def render(
    self,
    arguments: dict[str, Any] | None,
    context: Context[LifespanContextT, RequestT],
) -> list[Message]:
    """Render the prompt with arguments.

    Raises:
        ValueError: If required arguments are missing, or if rendering fails.
    """
    # Validate required arguments
    if self.arguments:
        required = {arg.name for arg in self.arguments if arg.required}
        provided = set(arguments or {})
        missing = required - provided
        if missing:
            raise ValueError(f"Missing required arguments: {missing}")

    try:
        # Add context to arguments if needed
        call_args = inject_context(self.fn, arguments or {}, context, self.context_kwarg)

        fn = self.fn
        if is_async_callable(fn):
            result = await fn(**call_args)
        else:
            result = await anyio.to_thread.run_sync(functools.partial(self.fn, **call_args))

        # Validate messages
        if not isinstance(result, list | tuple):
            result = [result]

        # Convert result to messages
        messages: list[Message] = []
        for msg in result:  # type: ignore[reportUnknownVariableType]
            try:
                if isinstance(msg, Message):
                    messages.append(msg)
                elif isinstance(msg, dict):
                    messages.append(message_validator.validate_python(msg))
                elif isinstance(msg, str):
                    content = TextContent(type="text", text=msg)
                    messages.append(UserMessage(content=content))
                else:  # pragma: no cover
                    content = pydantic_core.to_json(msg, fallback=str, indent=2).decode()
                    messages.append(Message(role="user", content=content))
            except Exception:  # pragma: no cover
                raise ValueError(f"Could not convert prompt result to message: {msg}")

        return messages
    except Exception as e:  # pragma: no cover
        raise ValueError(f"Error rendering prompt {self.name}: {e}")

PromptManager

Manages MCPServer prompts.

Source code in src/mcp/server/mcpserver/prompts/manager.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class PromptManager:
    """Manages MCPServer prompts."""

    def __init__(self, warn_on_duplicate_prompts: bool = True):
        self._prompts: dict[str, Prompt] = {}
        self.warn_on_duplicate_prompts = warn_on_duplicate_prompts

    def get_prompt(self, name: str) -> Prompt | None:
        """Get prompt by name."""
        return self._prompts.get(name)

    def list_prompts(self) -> list[Prompt]:
        """List all registered prompts."""
        return list(self._prompts.values())

    def add_prompt(
        self,
        prompt: Prompt,
    ) -> Prompt:
        """Add a prompt to the manager."""

        # Check for duplicates
        existing = self._prompts.get(prompt.name)
        if existing:
            if self.warn_on_duplicate_prompts:
                logger.warning(f"Prompt already exists: {prompt.name}")
            return existing

        self._prompts[prompt.name] = prompt
        return prompt

    async def render_prompt(
        self,
        name: str,
        arguments: dict[str, Any] | None,
        context: Context[LifespanContextT, RequestT],
    ) -> list[Message]:
        """Render a prompt by name with arguments."""
        prompt = self.get_prompt(name)
        if not prompt:
            raise ValueError(f"Unknown prompt: {name}")

        return await prompt.render(arguments, context)

get_prompt 

get_prompt(name: str) -> Prompt | None

Get prompt by name.

Source code in src/mcp/server/mcpserver/prompts/manager.py 
24
25
26
def get_prompt(self, name: str) -> Prompt | None:
    """Get prompt by name."""
    return self._prompts.get(name)

list_prompts 

list_prompts() -> list[Prompt]

List all registered prompts.

Source code in src/mcp/server/mcpserver/prompts/manager.py 
28
29
30
def list_prompts(self) -> list[Prompt]:
    """List all registered prompts."""
    return list(self._prompts.values())

add_prompt 

add_prompt(prompt: Prompt) -> Prompt

Add a prompt to the manager.

Source code in src/mcp/server/mcpserver/prompts/manager.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def add_prompt(
    self,
    prompt: Prompt,
) -> Prompt:
    """Add a prompt to the manager."""

    # Check for duplicates
    existing = self._prompts.get(prompt.name)
    if existing:
        if self.warn_on_duplicate_prompts:
            logger.warning(f"Prompt already exists: {prompt.name}")
        return existing

    self._prompts[prompt.name] = prompt
    return prompt

render_prompt async 

render_prompt(
    name: str,
    arguments: dict[str, Any] | None,
    context: Context[LifespanContextT, RequestT],
) -> list[Message]

Render a prompt by name with arguments.

Source code in src/mcp/server/mcpserver/prompts/manager.py 
48
49
50
51
52
53
54
55
56
57
58
59
async def render_prompt(
    self,
    name: str,
    arguments: dict[str, Any] | None,
    context: Context[LifespanContextT, RequestT],
) -> list[Message]:
    """Render a prompt by name with arguments."""
    prompt = self.get_prompt(name)
    if not prompt:
        raise ValueError(f"Unknown prompt: {name}")

    return await prompt.render(arguments, context)