feat(managers): add tenant-scoped storage to ToolManager, ResourceManager, and PromptManager

Change internal storage from simple name-keyed dicts to composite
(tenant_id, name) keyed dicts, enabling the same resource name to exist
independently under different tenants.

All public methods gain a keyword-only `tenant_id: str | None = None`
parameter. Existing callers are unaffected — the default None scope
preserves backward compatibility with single-tenant usage.

Author: andylim-duo

src/mcp/server/mcpserver/prompts/manager.py

@@ -15,44 +15,55 @@
 
 
 class PromptManager:
-    """Manages MCPServer prompts."""
+    """Manages MCPServer prompts with optional tenant-scoped storage.
+
+    Prompts are stored in a dict keyed by ``(tenant_id, prompt_name)``.
+    This allows the same prompt name to exist independently under different
+    tenants. When ``tenant_id`` is ``None`` (the default), prompts live in
+    a global scope, preserving backward compatibility with single-tenant usage.
+    """
 
     def __init__(self, warn_on_duplicate_prompts: bool = True):
-        self._prompts: dict[str, Prompt] = {}
+        self._prompts: dict[tuple[str | None, 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 get_prompt(self, name: str, *, tenant_id: str | None = None) -> Prompt | None:
+        """Get prompt by name, optionally scoped to a tenant."""
+        return self._prompts.get((tenant_id, name))
 
-    def list_prompts(self) -> list[Prompt]:
-        """List all registered prompts."""
-        return list(self._prompts.values())
+    def list_prompts(self, *, tenant_id: str | None = None) -> list[Prompt]:
+        """List all registered prompts for a given tenant scope."""
+        return [prompt for (tid, _), prompt in self._prompts.items() if tid == tenant_id]
 
     def add_prompt(
         self,
         prompt: Prompt,
+        *,
+        tenant_id: str | None = None,
     ) -> Prompt:
-        """Add a prompt to the manager."""
+        """Add a prompt to the manager, optionally scoped to a tenant."""
 
         # Check for duplicates
-        existing = self._prompts.get(prompt.name)
+        key = (tenant_id, prompt.name)
+        existing = self._prompts.get(key)
         if existing:
             if self.warn_on_duplicate_prompts:
                 logger.warning(f"Prompt already exists: {prompt.name}")
             return existing
 
-        self._prompts[prompt.name] = prompt
+        self._prompts[key] = prompt
         return prompt
 
     async def render_prompt(
         self,
         name: str,
         arguments: dict[str, Any] | None,
         context: Context[LifespanContextT, RequestT],
+        *,
+        tenant_id: str | None = None,
     ) -> list[Message]:
-        """Render a prompt by name with arguments."""
-        prompt = self.get_prompt(name)
+        """Render a prompt by name with arguments, optionally scoped to a tenant."""
+        prompt = self.get_prompt(name, tenant_id=tenant_id)
         if not prompt:
             raise ValueError(f"Unknown prompt: {name}")
 

src/mcp/server/mcpserver/resources/resource_manager.py

@@ -20,18 +20,27 @@
 
 
 class ResourceManager:
-    """Manages MCPServer resources."""
+    """Manages MCPServer resources with optional tenant-scoped storage.
+
+    Resources and templates are stored in dicts keyed by
+    ``(tenant_id, uri_string)`` and ``(tenant_id, uri_template)``
+    respectively. This allows the same URI to exist independently under
+    different tenants. When ``tenant_id`` is ``None`` (the default),
+    entries live in a global scope, preserving backward compatibility
+    with single-tenant usage.
+    """
 
     def __init__(self, warn_on_duplicate_resources: bool = True):
-        self._resources: dict[str, Resource] = {}
-        self._templates: dict[str, ResourceTemplate] = {}
+        self._resources: dict[tuple[str | None, str], Resource] = {}
+        self._templates: dict[tuple[str | None, str], ResourceTemplate] = {}
         self.warn_on_duplicate_resources = warn_on_duplicate_resources
 
-    def add_resource(self, resource: Resource) -> Resource:
-        """Add a resource to the manager.
+    def add_resource(self, resource: Resource, *, tenant_id: str | None = None) -> Resource:
+        """Add a resource to the manager, optionally scoped to a tenant.
 
         Args:
             resource: A Resource instance to add
+            tenant_id: Optional tenant scope for the resource
 
         Returns:
             The added resource. If a resource with the same URI already exists,
@@ -45,12 +54,13 @@ def add_resource(self, resource: Resource) -> Resource:
                 "resource_name": resource.name,
             },
         )
-        existing = self._resources.get(str(resource.uri))
+        key = (tenant_id, str(resource.uri))
+        existing = self._resources.get(key)
         if existing:
             if self.warn_on_duplicate_resources:
                 logger.warning(f"Resource already exists: {resource.uri}")
             return existing
-        self._resources[str(resource.uri)] = resource
+        self._resources[key] = resource
         return resource
 
     def add_template(
@@ -64,8 +74,10 @@ def add_template(
         icons: list[Icon] | None = None,
         annotations: Annotations | None = None,
         meta: dict[str, Any] | None = None,
+        *,
+        tenant_id: str | None = None,
     ) -> ResourceTemplate:
-        """Add a template from a function."""
+        """Add a template from a function, optionally scoped to a tenant."""
         template = ResourceTemplate.from_function(
             fn,
             uri_template=uri_template,
@@ -77,20 +89,28 @@ def add_template(
             annotations=annotations,
             meta=meta,
         )
-        self._templates[template.uri_template] = template
+        self._templates[(tenant_id, template.uri_template)] = template
         return template
 
-    async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
+    async def get_resource(
+        self,
+        uri: AnyUrl | str,
+        context: Context[LifespanContextT, RequestT],
+        *,
+        tenant_id: str | None = None,
+    ) -> Resource:
         """Get resource by URI, checking concrete resources first, then templates."""
         uri_str = str(uri)
         logger.debug("Getting resource", extra={"uri": uri_str})
 
         # First check concrete resources
-        if resource := self._resources.get(uri_str):
+        if resource := self._resources.get((tenant_id, uri_str)):
             return resource
 
-        # Then check templates
-        for template in self._templates.values():
+        # Then check templates for this tenant scope
+        for (tid, _), template in self._templates.items():
+            if tid != tenant_id:
+                continue
             if params := template.matches(uri_str):
                 try:
                     return await template.create_resource(uri_str, params, context=context)
@@ -99,12 +119,14 @@ async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContext
 
         raise ValueError(f"Unknown resource: {uri}")
 
-    def list_resources(self) -> list[Resource]:
-        """List all registered resources."""
-        logger.debug("Listing resources", extra={"count": len(self._resources)})
-        return list(self._resources.values())
-
-    def list_templates(self) -> list[ResourceTemplate]:
-        """List all registered templates."""
-        logger.debug("Listing templates", extra={"count": len(self._templates)})
-        return list(self._templates.values())
+    def list_resources(self, *, tenant_id: str | None = None) -> list[Resource]:
+        """List all registered resources for a given tenant scope."""
+        resources = [r for (tid, _), r in self._resources.items() if tid == tenant_id]
+        logger.debug("Listing resources", extra={"count": len(resources)})
+        return resources
+
+    def list_templates(self, *, tenant_id: str | None = None) -> list[ResourceTemplate]:
+        """List all registered templates for a given tenant scope."""
+        templates = [t for (tid, _), t in self._templates.items() if tid == tenant_id]
+        logger.debug("Listing templates", extra={"count": len(templates)})
+        return templates

src/mcp/server/mcpserver/tools/tool_manager.py

@@ -16,30 +16,37 @@
 
 
 class ToolManager:
-    """Manages MCPServer tools."""
+    """Manages MCPServer tools with optional tenant-scoped storage.
+
+    Tools are stored in a dict keyed by ``(tenant_id, tool_name)``.
+    This allows the same tool name to exist independently under different
+    tenants. When ``tenant_id`` is ``None`` (the default), tools live in
+    a global scope, preserving backward compatibility with single-tenant usage.
+    """
 
     def __init__(
         self,
         warn_on_duplicate_tools: bool = True,
         *,
         tools: list[Tool] | None = None,
     ):
-        self._tools: dict[str, Tool] = {}
+        self._tools: dict[tuple[str | None, str], Tool] = {}
         if tools is not None:
             for tool in tools:
-                if warn_on_duplicate_tools and tool.name in self._tools:
+                key = (None, tool.name)
+                if warn_on_duplicate_tools and key in self._tools:
                     logger.warning(f"Tool already exists: {tool.name}")
-                self._tools[tool.name] = tool
+                self._tools[key] = tool
 
         self.warn_on_duplicate_tools = warn_on_duplicate_tools
 
-    def get_tool(self, name: str) -> Tool | None:
-        """Get tool by name."""
-        return self._tools.get(name)
+    def get_tool(self, name: str, *, tenant_id: str | None = None) -> Tool | None:
+        """Get tool by name, optionally scoped to a tenant."""
+        return self._tools.get((tenant_id, name))
 
-    def list_tools(self) -> list[Tool]:
-        """List all registered tools."""
-        return list(self._tools.values())
+    def list_tools(self, *, tenant_id: str | None = None) -> list[Tool]:
+        """List all registered tools for a given tenant scope."""
+        return [tool for (tid, _), tool in self._tools.items() if tid == tenant_id]
 
     def add_tool(
         self,
@@ -51,8 +58,10 @@ def add_tool(
         icons: list[Icon] | None = None,
         meta: dict[str, Any] | None = None,
         structured_output: bool | None = None,
+        *,
+        tenant_id: str | None = None,
     ) -> Tool:
-        """Add a tool to the server."""
+        """Add a tool to the server, optionally scoped to a tenant."""
         tool = Tool.from_function(
             fn,
             name=name,
@@ -63,29 +72,33 @@ def add_tool(
             meta=meta,
             structured_output=structured_output,
         )
-        existing = self._tools.get(tool.name)
+        key = (tenant_id, tool.name)
+        existing = self._tools.get(key)
         if existing:
             if self.warn_on_duplicate_tools:
                 logger.warning(f"Tool already exists: {tool.name}")
             return existing
-        self._tools[tool.name] = tool
+        self._tools[key] = tool
         return tool
 
-    def remove_tool(self, name: str) -> None:
-        """Remove a tool by name."""
-        if name not in self._tools:
+    def remove_tool(self, name: str, *, tenant_id: str | None = None) -> None:
+        """Remove a tool by name, optionally scoped to a tenant."""
+        key = (tenant_id, name)
+        if key not in self._tools:
             raise ToolError(f"Unknown tool: {name}")
-        del self._tools[name]
+        del self._tools[key]
 
     async def call_tool(
         self,
         name: str,
         arguments: dict[str, Any],
         context: Context[LifespanContextT, RequestT],
         convert_result: bool = False,
+        *,
+        tenant_id: str | None = None,
     ) -> Any:
-        """Call a tool by name with arguments."""
-        tool = self.get_tool(name)
+        """Call a tool by name with arguments, optionally scoped to a tenant."""
+        tool = self.get_tool(name, tenant_id=tenant_id)
         if not tool:
             raise ToolError(f"Unknown tool: {name}")
 

tests/server/mcpserver/resources/test_resource_manager.py

@@ -103,7 +103,7 @@ def greet(name: str) -> str:
             uri_template="greet://{name}",
             name="greeter",
         )
-        manager._templates[template.uri_template] = template
+        manager._templates[(None, template.uri_template)] = template
 
         resource = await manager.get_resource(AnyUrl("greet://world"), Context())
         assert isinstance(resource, FunctionResource)