Skip to content
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
4 changes: 4 additions & 0 deletions src/mcp/server/mcpserver/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,5 +17,9 @@ class ToolError(MCPServerError):
"""Error in tool operations."""


class PromptError(MCPServerError):
"""Error in prompt operations."""


class InvalidSignature(Exception):
"""Invalid signature for use with MCPServer."""
7 changes: 7 additions & 0 deletions src/mcp/server/mcpserver/prompts/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from typing import TYPE_CHECKING, Any

from mcp.server.mcpserver.exceptions import PromptError
from mcp.server.mcpserver.prompts.base import Message, Prompt
from mcp.server.mcpserver.utilities.logging import get_logger

Expand Down Expand Up @@ -45,6 +46,12 @@ def add_prompt(
self._prompts[prompt.name] = prompt
return prompt

def remove_prompt(self, name: str) -> None:
"""Remove a prompt by name."""
if name not in self._prompts:
raise PromptError(f"Unknown prompt: {name}")
del self._prompts[name]

async def render_prompt(
self,
name: str,
Expand Down
28 changes: 28 additions & 0 deletions src/mcp/server/mcpserver/resources/resource_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from pydantic import AnyUrl

from mcp.server.mcpserver.exceptions import ResourceError
from mcp.server.mcpserver.resources.base import Resource
from mcp.server.mcpserver.resources.templates import ResourceTemplate
from mcp.server.mcpserver.utilities.logging import get_logger
Expand Down Expand Up @@ -53,6 +54,20 @@ def add_resource(self, resource: Resource) -> Resource:
self._resources[str(resource.uri)] = resource
return resource

def remove_resource(self, uri: AnyUrl | str) -> None:
"""Remove a resource by URI.

Args:
uri: The URI of the resource to remove

Raises:
ResourceError: If the resource does not exist
"""
uri_str = str(uri)
if uri_str not in self._resources:
raise ResourceError(f"Unknown resource: {uri}")
del self._resources[uri_str]

def add_template(
self,
fn: Callable[..., Any],
Expand Down Expand Up @@ -80,6 +95,19 @@ def add_template(
self._templates[template.uri_template] = template
return template

def remove_template(self, uri_template: str) -> None:
"""Remove a resource template by URI template.

Args:
uri_template: The URI template string to remove

Raises:
ResourceError: If the template does not exist
"""
if uri_template not in self._templates:
raise ResourceError(f"Unknown resource template: {uri_template}")
del self._templates[uri_template]

async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
"""Get resource by URI, checking concrete resources first, then templates."""
uri_str = str(uri)
Expand Down
33 changes: 33 additions & 0 deletions src/mcp/server/mcpserver/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -615,6 +615,28 @@ def add_resource(self, resource: Resource) -> None:
"""
self._resource_manager.add_resource(resource)

def remove_resource(self, uri: str) -> None:
"""Remove a resource from the server by URI.

Args:
uri: The URI of the resource to remove

Raises:
ResourceError: If the resource does not exist
"""
self._resource_manager.remove_resource(uri)

def remove_resource_template(self, uri_template: str) -> None:
"""Remove a resource template from the server by URI template.

Args:
uri_template: The URI template string to remove

Raises:
ResourceError: If the template does not exist
"""
self._resource_manager.remove_template(uri_template)

def resource(
self,
uri: str,
Expand Down Expand Up @@ -735,6 +757,17 @@ def add_prompt(self, prompt: Prompt) -> None:
"""
self._prompt_manager.add_prompt(prompt)

def remove_prompt(self, name: str) -> None:
"""Remove a prompt from the server by name.

Args:
name: The name of the prompt to remove

Raises:
PromptError: If the prompt does not exist
"""
self._prompt_manager.remove_prompt(name)

def prompt(
self,
name: str | None = None,
Expand Down
171 changes: 170 additions & 1 deletion tests/server/mcpserver/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from mcp.server.context import ServerRequestContext
from mcp.server.experimental.request_context import Experimental
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.mcpserver.exceptions import PromptError, ResourceError, ToolError
from mcp.server.mcpserver.prompts.base import Message, UserMessage
from mcp.server.mcpserver.resources import FileResource, FunctionResource
from mcp.server.mcpserver.utilities.types import Audio, Image
Expand Down Expand Up @@ -785,6 +785,69 @@ def get_data() -> str: # pragma: no cover
assert resource.name == "test_get_data"
assert resource.mime_type == "text/plain"

async def test_remove_resource(self):
"""Test removing a resource from the server."""
mcp = MCPServer()

@mcp.resource("resource://test")
def get_data() -> str: # pragma: no cover
return "Hello"

assert len(mcp._resource_manager.list_resources()) == 1

mcp.remove_resource("resource://test")

assert len(mcp._resource_manager.list_resources()) == 0

async def test_remove_nonexistent_resource(self):
"""Test that removing a non-existent resource raises ResourceError."""
mcp = MCPServer()

with pytest.raises(ResourceError, match="Unknown resource: resource://nonexistent"):
mcp.remove_resource("resource://nonexistent")

async def test_remove_resource_and_list(self):
"""Test that a removed resource doesn't appear in list_resources."""
mcp = MCPServer()

@mcp.resource("resource://first")
def first() -> str: # pragma: no cover
return "first"

@mcp.resource("resource://second")
def second() -> str: # pragma: no cover
return "second"

async with Client(mcp) as client:
resources = await client.list_resources()
assert len(resources.resources) == 2

mcp.remove_resource("resource://first")

async with Client(mcp) as client:
resources = await client.list_resources()
assert len(resources.resources) == 1
assert resources.resources[0].uri == "resource://second"

async def test_remove_resource_and_read(self):
"""Test that reading a removed resource fails appropriately."""
mcp = MCPServer()

@mcp.resource("resource://test")
def get_data() -> str:
return "Hello"

async with Client(mcp) as client:
result = await client.read_resource("resource://test")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Hello"

mcp.remove_resource("resource://test")

async with Client(mcp) as client:
with pytest.raises(MCPError, match="Unknown resource"):
await client.read_resource("resource://test")


class TestServerResourceTemplates:
async def test_resource_with_params(self):
Expand Down Expand Up @@ -920,6 +983,50 @@ def get_csv(user: str) -> str:
)
)

async def test_remove_resource_template(self):
"""Test removing a resource template from the server."""
mcp = MCPServer()

@mcp.resource("resource://{name}/data")
def get_data(name: str) -> str: # pragma: no cover
return f"Data for {name}"

assert len(mcp._resource_manager._templates) == 1

mcp.remove_resource_template("resource://{name}/data")

assert len(mcp._resource_manager._templates) == 0

async def test_remove_nonexistent_resource_template(self):
"""Test that removing a non-existent template raises ResourceError."""
mcp = MCPServer()

with pytest.raises(ResourceError, match="Unknown resource template: resource://\\{name\\}/data"):
mcp.remove_resource_template("resource://{name}/data")

async def test_remove_resource_template_and_list(self):
"""Test that a removed template doesn't appear in list_resource_templates."""
mcp = MCPServer()

@mcp.resource("resource://{name}/first")
def first(name: str) -> str: # pragma: no cover
return f"first {name}"

@mcp.resource("resource://{name}/second")
def second(name: str) -> str: # pragma: no cover
return f"second {name}"

async with Client(mcp) as client:
templates = await client.list_resource_templates()
assert len(templates.resource_templates) == 2

mcp.remove_resource_template("resource://{name}/first")

async with Client(mcp) as client:
templates = await client.list_resource_templates()
assert len(templates.resource_templates) == 1
assert templates.resource_templates[0].uri_template == "resource://{name}/second"


class TestServerResourceMetadata:
"""Test MCPServer @resource decorator meta parameter for list operations.
Expand Down Expand Up @@ -1418,6 +1525,68 @@ def prompt_fn(name: str) -> str: ... # pragma: no branch
with pytest.raises(MCPError, match="Missing required arguments"):
await client.get_prompt("prompt_fn")

async def test_remove_prompt(self):
"""Test removing a prompt from the server."""
mcp = MCPServer()

@mcp.prompt()
def fn() -> str: # pragma: no cover
return "Hello"

assert len(mcp._prompt_manager.list_prompts()) == 1

mcp.remove_prompt("fn")

assert len(mcp._prompt_manager.list_prompts()) == 0

async def test_remove_nonexistent_prompt(self):
"""Test that removing a non-existent prompt raises PromptError."""
mcp = MCPServer()

with pytest.raises(PromptError, match="Unknown prompt: nonexistent"):
mcp.remove_prompt("nonexistent")

async def test_remove_prompt_and_list(self):
"""Test that a removed prompt doesn't appear in list_prompts."""
mcp = MCPServer()

@mcp.prompt()
def first() -> str: # pragma: no cover
return "first"

@mcp.prompt()
def second() -> str: # pragma: no cover
return "second"

async with Client(mcp) as client:
prompts = await client.list_prompts()
assert len(prompts.prompts) == 2

mcp.remove_prompt("first")

async with Client(mcp) as client:
prompts = await client.list_prompts()
assert len(prompts.prompts) == 1
assert prompts.prompts[0].name == "second"

async def test_remove_prompt_and_get(self):
"""Test that getting a removed prompt fails appropriately."""
mcp = MCPServer()

@mcp.prompt()
def fn() -> str:
return "Hello"

async with Client(mcp) as client:
result = await client.get_prompt("fn")
assert result.messages[0].content == TextContent(text="Hello")

mcp.remove_prompt("fn")

async with Client(mcp) as client:
with pytest.raises(MCPError, match="Unknown prompt"):
await client.get_prompt("fn")


async def test_completion_decorator() -> None:
"""Test that the completion decorator registers a working handler."""
Expand Down
Loading