Skip to content

SEP: Namespaces using URIs #1230

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 15 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
6 changes: 3 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -1125,7 +1125,7 @@ server = Server("example-server", lifespan=server_lifespan)


@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
async def handle_list_tools(_) -> list[types.Tool]:
"""List available tools."""
return [
types.Tool(
Expand Down Expand Up @@ -1207,7 +1207,7 @@ server = Server("example-server")


@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
async def handle_list_prompts(_) -> list[types.Prompt]:
"""List available prompts."""
return [
types.Prompt(
Expand Down Expand Up @@ -1286,7 +1286,7 @@ server = Server("example-server")


@server.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
"""List available tools with structured output schemas."""
return [
types.Tool(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ def main(port: int, transport: str) -> int:
app = Server("mcp-simple-prompt")

@app.list_prompts()
async def list_prompts() -> list[types.Prompt]:
async def list_prompts(_) -> list[types.Prompt]:
return [
types.Prompt(
name="simple",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ def main(port: int, transport: str) -> int:
app = Server("mcp-simple-resource")

@app.list_resources()
async def list_resources() -> list[types.Resource]:
async def list_resources(_) -> list[types.Resource]:
return [
types.Resource(
uri=FileUrl(f"file:///{name}.txt"),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
]

@app.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
return [
types.Tool(
name="start-notification-stream",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
]

@app.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
return [
types.Tool(
name="start-notification-stream",
Expand Down
2 changes: 1 addition & 1 deletion examples/servers/simple-tool/mcp_simple_tool/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ async def fetch_tool(name: str, arguments: dict) -> list[types.ContentBlock]:
return await fetch_website(arguments["url"])

@app.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
return [
types.Tool(
name="fetch",
Expand Down
2 changes: 1 addition & 1 deletion examples/servers/structured_output_lowlevel.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@


@server.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
"""List available tools with their schemas."""
return [
types.Tool(
Expand Down
182 changes: 182 additions & 0 deletions examples/snippets/servers/hierarchical_organization.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
"""Example demonstrating hierarchical organization of tools, prompts, and resources using custom URIs.

This example shows how to:
1. Register tools, prompts, and resources with hierarchical URIs
2. Create group discovery resources at well-known URIs
3. Filter items by URI paths for better organization
"""

import json
from typing import cast

from pydantic import AnyUrl

from mcp.server.fastmcp import FastMCP
from mcp.types import ListFilters, TextContent, TextResourceContents

# Create FastMCP server instance
mcp = FastMCP("hierarchical-example")


# Group discovery resources
@mcp.resource("mcp://groups/tools")
def get_tool_groups() -> str:
"""Discover available tool groups."""
return json.dumps(
{
"groups": [
{"name": "math", "description": "Mathematical operations", "uri_paths": ["mcp://tools/math/"]},
{"name": "string", "description": "String manipulation", "uri_paths": ["mcp://tools/string/"]},
]
},
indent=2,
)


@mcp.resource("mcp://groups/prompts")
def get_prompt_groups() -> str:
"""Discover available prompt groups."""
return json.dumps(
{
"groups": [
{"name": "greetings", "description": "Greeting prompts", "uri_paths": ["mcp://prompts/greetings/"]},
{
"name": "instructions",
"description": "Instructional prompts",
"uri_paths": ["mcp://prompts/instructions/"],
},
]
},
indent=2,
)


# Math tools organized under mcp://tools/math/
@mcp.tool(uri="mcp://tools/math/add")
def add(a: float, b: float) -> float:
"""Add two numbers."""
return a + b


@mcp.tool(uri="mcp://tools/math/multiply")
def multiply(a: float, b: float) -> float:
"""Multiply two numbers."""
return a * b


# String tools organized under mcp://tools/string/
@mcp.tool(uri="mcp://tools/string/reverse")
def reverse(text: str) -> str:
"""Reverse a string."""
return text[::-1]


@mcp.tool(uri="mcp://tools/string/upper")
def upper(text: str) -> str:
"""Convert to uppercase."""
return text.upper()


# Greeting prompts organized under mcp://prompts/greetings/
@mcp.prompt(uri="mcp://prompts/greetings/hello")
def hello_prompt(name: str) -> str:
"""Generate a hello greeting."""
return f"Hello, {name}! How can I help you today?"


@mcp.prompt(uri="mcp://prompts/greetings/goodbye")
def goodbye_prompt(name: str) -> str:
"""Generate a goodbye message."""
return f"Goodbye, {name}! Have a great day!"


# Instruction prompts organized under mcp://prompts/instructions/
@mcp.prompt(uri="mcp://prompts/instructions/setup")
def setup_prompt(tool: str) -> str:
"""Generate setup instructions for a tool."""
return (
f"To set up {tool}, follow these steps:\n"
"1. Install the required dependencies\n"
"2. Configure the settings\n"
"3. Run the initialization script\n"
"4. Verify the installation"
)


@mcp.prompt(uri="mcp://prompts/instructions/debug")
def debug_prompt(error: str) -> str:
"""Generate debugging instructions for an error."""
return (
f"To debug '{error}':\n"
"1. Check the error logs\n"
"2. Verify input parameters\n"
"3. Enable verbose logging\n"
"4. Isolate the issue with minimal reproduction"
)


if __name__ == "__main__":
# Example of testing the hierarchical organization
import asyncio

from mcp.shared.memory import create_connected_server_and_client_session

async def test_hierarchy():
"""Test the hierarchical organization."""
async with create_connected_server_and_client_session(mcp._mcp_server) as client:
# 1. Discover tool groups and list tools in each group
print("\n=== Discovering Tool Groups ===")
result = await client.read_resource(uri=AnyUrl("mcp://groups/tools"))
tool_groups = json.loads(cast(TextResourceContents, result.contents[0]).text)

for group in tool_groups["groups"]:
print(f"\n--- {group['name'].upper()} Tools ({group['description']}) ---")
# Use the URI paths from the group definition
group_tools = await client.list_tools(
filters=ListFilters(uri_paths=[AnyUrl(uri) for uri in group["uri_paths"]])
)
for tool in group_tools.tools:
print(f" - {tool.name}: {tool.description}")

# 2. Call tools by name (still works!)
print("\n=== Calling Tools by Name ===")
result = await client.call_tool("add", {"a": 10, "b": 5})
print(f"add(10, 5) = {cast(TextContent, result.content[0]).text}")

result = await client.call_tool("reverse", {"text": "Hello"})
print(f"reverse('Hello') = {cast(TextContent, result.content[0]).text}")

# 3. Call tools by URI
print("\n=== Calling Tools by URI ===")
result = await client.call_tool("mcp://tools/math/multiply", {"a": 7, "b": 8})
print(
f"Call mcp://tools/math/multiply with {{'a': 7, 'b': 8}} = {cast(TextContent, result.content[0]).text}"
)

result = await client.call_tool("mcp://tools/string/upper", {"text": "hello world"})
print(
f"Call mcp://tools/string/upper with {{'text': 'hello world'}} = "
f"{cast(TextContent, result.content[0]).text}"
)

# 4. Discover prompt groups and list prompts in each group
print("\n=== Discovering Prompt Groups ===")
result = await client.read_resource(uri=AnyUrl("mcp://groups/prompts"))
prompt_groups = json.loads(cast(TextResourceContents, result.contents[0]).text)

for group in prompt_groups["groups"]:
print(f"\n--- {group['name'].upper()} Prompts ({group['description']}) ---")
# Use the URI paths from the group definition
group_prompts = await client.list_prompts(
filters=ListFilters(uri_paths=[AnyUrl(uri) for uri in group["uri_paths"]])
)
for prompt in group_prompts.prompts:
print(f" - {prompt.name}: {prompt.description}")

# 5. Use a prompt
print("\n=== Using a Prompt ===")
result = await client.get_prompt("hello_prompt", {"name": "Alice"})
print(f"Prompt result: {cast(TextContent, result.messages[0].content).text}")

# Run the test
asyncio.run(test_hierarchy())
2 changes: 1 addition & 1 deletion examples/snippets/servers/lowlevel/basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@


@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
async def handle_list_prompts(_) -> list[types.Prompt]:
"""List available prompts."""
return [
types.Prompt(
Expand Down
2 changes: 1 addition & 1 deletion examples/snippets/servers/lowlevel/lifespan.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ async def server_lifespan(_server: Server) -> AsyncIterator[dict]:


@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
async def handle_list_tools(_) -> list[types.Tool]:
"""List available tools."""
return [
types.Tool(
Expand Down
2 changes: 1 addition & 1 deletion examples/snippets/servers/lowlevel/structured_output.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@


@server.list_tools()
async def list_tools() -> list[types.Tool]:
async def list_tools(_) -> list[types.Tool]:
"""List available tools with structured output schemas."""
return [
types.Tool(
Expand Down
38 changes: 30 additions & 8 deletions src/mcp/client/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -221,25 +221,37 @@ async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResul
types.EmptyResult,
)

async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult:
async def list_resources(
self, filters: types.ListFilters | None = None, cursor: str | None = None
) -> types.ListResourcesResult:
"""Send a resources/list request."""
params = None
if cursor is not None or filters is not None:
params = types.ListRequestParams(filters=filters, cursor=cursor)
return await self.send_request(
types.ClientRequest(
types.ListResourcesRequest(
method="resources/list",
params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
params=params,
)
),
types.ListResourcesResult,
)

async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult:
async def list_resource_templates(
self,
filters: types.ListFilters | None = None,
cursor: str | None = None,
) -> types.ListResourceTemplatesResult:
"""Send a resources/templates/list request."""
params = None
if cursor is not None or filters is not None:
params = types.ListRequestParams(filters=filters, cursor=cursor)
return await self.send_request(
types.ClientRequest(
types.ListResourceTemplatesRequest(
method="resources/templates/list",
params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
params=params,
)
),
types.ListResourceTemplatesResult,
Expand Down Expand Up @@ -332,13 +344,18 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) -
except SchemaError as e:
raise RuntimeError(f"Invalid schema for tool {name}: {e}")

async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult:
async def list_prompts(
self, filters: types.ListFilters | None = None, cursor: str | None = None
) -> types.ListPromptsResult:
"""Send a prompts/list request."""
params = None
if cursor is not None or filters is not None:
params = types.ListRequestParams(filters=filters, cursor=cursor)
return await self.send_request(
types.ClientRequest(
types.ListPromptsRequest(
method="prompts/list",
params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
params=params,
)
),
types.ListPromptsResult,
Expand Down Expand Up @@ -381,13 +398,18 @@ async def complete(
types.CompleteResult,
)

async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult:
async def list_tools(
self, filters: types.ListFilters | None = None, cursor: str | None = None
) -> types.ListToolsResult:
"""Send a tools/list request."""
params = None
if cursor is not None or filters is not None:
params = types.ListRequestParams(filters=filters, cursor=cursor)
result = await self.send_request(
types.ClientRequest(
types.ListToolsRequest(
method="tools/list",
params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
params=params,
)
),
types.ListToolsResult,
Expand Down
Loading
Loading