Fix/issue_1574

README.md

@@ -1032,7 +1032,7 @@ async def notify_data_update(resource_uri: str, ctx: Context) -> str:
     # Perform data update logic here
 
     # Notify clients that this specific resource changed
-    await ctx.session.send_resource_updated(AnyUrl(resource_uri))
+    await ctx.session.send_resource_updated(resource_uri)
 
     # If this affects the overall resource list, notify about that too
     await ctx.session.send_resource_list_changed()
@@ -1923,8 +1923,6 @@ For servers that need to handle large datasets, the low-level server provides pa
 Example of implementing pagination with MCP server decorators.
 """
 
-from pydantic import AnyUrl
-
 import mcp.types as types
 from mcp.server.lowlevel import Server
 
@@ -1949,7 +1947,7 @@ async def list_resources_paginated(request: types.ListResourcesRequest) -> types
 
     # Get page of resources
     page_items = [
-        types.Resource(uri=AnyUrl(f"resource://items/{item}"), name=item, description=f"Description for {item}")
+        types.Resource(uri=f"resource://items/{item}", name=item, description=f"Description for {item}")
         for item in ITEMS[start:end]
     ]
 
@@ -2035,8 +2033,6 @@ cd to the `examples/snippets/clients` directory and run:
 import asyncio
 import os
 
-from pydantic import AnyUrl
-
 from mcp import ClientSession, StdioServerParameters, types
 from mcp.client.stdio import stdio_client
 from mcp.shared.context import RequestContext
@@ -2089,7 +2085,7 @@ async def run():
             print(f"Available tools: {[t.name for t in tools.tools]}")
 
             # Read a resource (greeting resource from fastmcp_quickstart)
-            resource_content = await session.read_resource(AnyUrl("greeting://World"))
+            resource_content = await session.read_resource("greeting://World")
             content_block = resource_content.contents[0]
             if isinstance(content_block, types.TextContent):
                 print(f"Resource content: {content_block.text}")
@@ -2256,8 +2252,6 @@ cd to the `examples/snippets` directory and run:
 import asyncio
 from urllib.parse import parse_qs, urlparse
 
-from pydantic import AnyUrl
-
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
 from mcp.client.streamable_http import streamablehttp_client
@@ -2304,7 +2298,7 @@ async def main():
         server_url="http://localhost:8001",
         client_metadata=OAuthClientMetadata(
             client_name="Example MCP Client",
-            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+            redirect_uris=["http://localhost:3000/callback"],
             grant_types=["authorization_code", "refresh_token"],
             response_types=["code"],
             scope="user",

examples/servers/simple-pagination/mcp_simple_pagination/server.py

@@ -11,7 +11,6 @@
 import click
 import mcp.types as types
 from mcp.server.lowlevel import Server
-from pydantic import AnyUrl
 from starlette.requests import Request
 
 # Sample data - in real scenarios, this might come from a database
@@ -27,7 +26,7 @@
 
 SAMPLE_RESOURCES = [
     types.Resource(
-        uri=AnyUrl(f"file:///path/to/resource_{i}.txt"),
+        uri=f"file:///path/to/resource_{i}.txt",
         name=f"resource_{i}",
         description=f"This is sample resource number {i}",
     )
@@ -160,7 +159,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentB
 
     # Implement read_resource handler
     @app.read_resource()
-    async def read_resource(uri: AnyUrl) -> str:
+    async def read_resource(uri: str) -> str:
         # Find the resource in our sample data
         resource = next((r for r in SAMPLE_RESOURCES if r.uri == uri), None)
         if not resource:

examples/servers/simple-resource/README.md

@@ -22,7 +22,6 @@ Using the MCP client, you can retrieve resources like this using the STDIO trans
 
 ```python
 import asyncio
-from mcp.types import AnyUrl
 from mcp.client.session import ClientSession
 from mcp.client.stdio import StdioServerParameters, stdio_client
 
@@ -39,7 +38,7 @@ async def main():
             print(resources)
 
             # Get a specific resource
-            resource = await session.read_resource(AnyUrl("file:///greeting.txt"))
+            resource = await session.read_resource("file:///greeting.txt")
             print(resource)
 
 

examples/servers/simple-resource/mcp_simple_resource/server.py

@@ -3,7 +3,6 @@
 import mcp.types as types
 from mcp.server.lowlevel import Server
 from mcp.server.lowlevel.helper_types import ReadResourceContents
-from pydantic import AnyUrl, FileUrl
 from starlette.requests import Request
 
 SAMPLE_RESOURCES = {
@@ -37,7 +36,7 @@ def main(port: int, transport: str) -> int:
     async def list_resources() -> list[types.Resource]:
         return [
             types.Resource(
-                uri=FileUrl(f"file:///{name}.txt"),
+                uri=f"file:///{name}.txt",
                 name=name,
                 title=SAMPLE_RESOURCES[name]["title"],
                 description=f"A sample text resource named {name}",
@@ -47,7 +46,7 @@ async def list_resources() -> list[types.Resource]:
         ]
 
     @app.read_resource()
-    async def read_resource(uri: AnyUrl):
+    async def read_resource(uri: str):
         if uri.path is None:
             raise ValueError(f"Invalid resource path: {uri}")
         name = uri.path.replace(".txt", "").lstrip("/")

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

@@ -8,7 +8,6 @@
 import mcp.types as types
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-from pydantic import AnyUrl
 from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
 from starlette.routing import Mount
@@ -74,7 +73,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentB
 
         # This will send a resource notificaiton though standalone SSE
         # established by GET request
-        await ctx.session.send_resource_updated(uri=AnyUrl("http:///test_resource"))
+        await ctx.session.send_resource_updated(uri="http:///test_resource")
         return [
             types.TextContent(
                 type="text",

examples/snippets/clients/oauth_client.py

@@ -10,8 +10,6 @@
 import asyncio
 from urllib.parse import parse_qs, urlparse
 
-from pydantic import AnyUrl
-
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
 from mcp.client.streamable_http import streamablehttp_client
@@ -58,7 +56,7 @@ async def main():
         server_url="http://localhost:8001",
         client_metadata=OAuthClientMetadata(
             client_name="Example MCP Client",
-            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+            redirect_uris=["http://localhost:3000/callback"],
             grant_types=["authorization_code", "refresh_token"],
             response_types=["code"],
             scope="user",

examples/snippets/clients/stdio_client.py

@@ -6,8 +6,6 @@
 import asyncio
 import os
 
-from pydantic import AnyUrl
-
 from mcp import ClientSession, StdioServerParameters, types
 from mcp.client.stdio import stdio_client
 from mcp.shared.context import RequestContext
@@ -60,7 +58,7 @@ async def run():
             print(f"Available tools: {[t.name for t in tools.tools]}")
 
             # Read a resource (greeting resource from fastmcp_quickstart)
-            resource_content = await session.read_resource(AnyUrl("greeting://World"))
+            resource_content = await session.read_resource("greeting://World")
             content_block = resource_content.contents[0]
             if isinstance(content_block, types.TextContent):
                 print(f"Resource content: {content_block.text}")

examples/snippets/servers/pagination_example.py

@@ -2,8 +2,6 @@
 Example of implementing pagination with MCP server decorators.
 """
 
-from pydantic import AnyUrl
-
 import mcp.types as types
 from mcp.server.lowlevel import Server
 
@@ -28,7 +26,7 @@ async def list_resources_paginated(request: types.ListResourcesRequest) -> types
 
     # Get page of resources
     page_items = [
-        types.Resource(uri=AnyUrl(f"resource://items/{item}"), name=item, description=f"Description for {item}")
+        types.Resource(uri=f"resource://items/{item}", name=item, description=f"Description for {item}")
         for item in ITEMS[start:end]
     ]
 

src/mcp/client/session.py

@@ -5,7 +5,7 @@
 import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from jsonschema import SchemaError, ValidationError, validate
-from pydantic import AnyUrl, TypeAdapter
+from pydantic import TypeAdapter
 from typing_extensions import deprecated
 
 import mcp.types as types
@@ -299,7 +299,7 @@ async def list_resource_templates(
             types.ListResourceTemplatesResult,
         )
 
-    async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
+    async def read_resource(self, uri: str) -> types.ReadResourceResult:
         """Send a resources/read request."""
         return await self.send_request(
             types.ClientRequest(
@@ -310,7 +310,7 @@ async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
             types.ReadResourceResult,
         )
 
-    async def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
+    async def subscribe_resource(self, uri: str) -> types.EmptyResult:
         """Send a resources/subscribe request."""
         return await self.send_request(
             types.ClientRequest(
@@ -321,7 +321,7 @@ async def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
             types.EmptyResult,
         )
 
-    async def unsubscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
+    async def unsubscribe_resource(self, uri: str) -> types.EmptyResult:
         """Send a resources/unsubscribe request."""
         return await self.send_request(
             types.ClientRequest(

src/mcp/server/auth/handlers/authorize.py

@@ -2,7 +2,7 @@
 from dataclasses import dataclass
 from typing import Any, Literal
 
-from pydantic import AnyUrl, BaseModel, Field, RootModel, ValidationError
+from pydantic import BaseModel, Field, RootModel, ValidationError
 from starlette.datastructures import FormData, QueryParams
 from starlette.requests import Request
 from starlette.responses import RedirectResponse, Response
@@ -24,7 +24,7 @@
 class AuthorizationRequest(BaseModel):
     # See https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.1
     client_id: str = Field(..., description="The client ID")
-    redirect_uri: AnyUrl | None = Field(None, description="URL to redirect to after authorization")
+    redirect_uri: str | None = Field(None, description="URL to redirect to after authorization")
 
     # see OAuthClientMetadata; we only support `code`
     response_type: Literal["code"] = Field(..., description="Must be 'code' for authorization code flow")
@@ -44,7 +44,7 @@ class AuthorizationRequest(BaseModel):
 class AuthorizationErrorResponse(BaseModel):
     error: AuthorizationErrorCode
     error_description: str | None
-    error_uri: AnyUrl | None = None
+    error_uri: str | None = None
     # must be set if provided in the request
     state: str | None = None
 
@@ -106,9 +106,7 @@ async def error_response(
                     if params is not None and "redirect_uri" not in params:
                         raw_redirect_uri = None
                     else:
-                        raw_redirect_uri = AnyUrlModel.model_validate(
-                            best_effort_extract_string("redirect_uri", params)
-                        ).root
+                        raw_redirect_uri = best_effort_extract_string("redirect_uri", params)
                     redirect_uri = client.validate_redirect_uri(raw_redirect_uri)
                 except (ValidationError, InvalidRedirectUriError):
                     # if the redirect URI is invalid, ignore it & just return the

src/mcp/server/auth/provider.py

@@ -2,7 +2,7 @@
 from typing import Generic, Literal, Protocol, TypeVar
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
-from pydantic import AnyUrl, BaseModel
+from pydantic import BaseModel
 
 from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
 
@@ -11,7 +11,7 @@ class AuthorizationParams(BaseModel):
     state: str | None
     scopes: list[str] | None
     code_challenge: str
-    redirect_uri: AnyUrl
+    redirect_uri: str
     redirect_uri_provided_explicitly: bool
     resource: str | None = None  # RFC 8707 resource indicator
 
@@ -22,7 +22,7 @@ class AuthorizationCode(BaseModel):
     expires_at: float
     client_id: str
     code_challenge: str
-    redirect_uri: AnyUrl
+    redirect_uri: str
     redirect_uri_provided_explicitly: bool
     resource: str | None = None  # RFC 8707 resource indicator
 

src/mcp/server/fastmcp/resources/base.py

@@ -4,7 +4,6 @@
 from typing import Annotated
 
 from pydantic import (
-    AnyUrl,
     BaseModel,
     ConfigDict,
     Field,
@@ -21,7 +20,7 @@ class Resource(BaseModel, abc.ABC):
 
     model_config = ConfigDict(validate_default=True)
 
-    uri: Annotated[AnyUrl, UrlConstraints(host_required=False)] = Field(default=..., description="URI of the resource")
+    uri: Annotated[str, UrlConstraints(host_required=False)] = Field(default=..., description="URI of the resource")
     name: str | None = Field(description="Name of the resource", default=None)
     title: str | None = Field(description="Human-readable title of the resource", default=None)
     description: str | None = Field(description="Description of the resource", default=None)

src/mcp/server/session.py

@@ -43,7 +43,6 @@ async def handle_list_prompts(ctx: RequestContext) -> list[types.Prompt]:
 import anyio
 import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
-from pydantic import AnyUrl
 
 import mcp.types as types
 from mcp.server.models import InitializationOptions
@@ -202,7 +201,7 @@ async def send_log_message(
             related_request_id,
         )
 
-    async def send_resource_updated(self, uri: AnyUrl) -> None:
+    async def send_resource_updated(self, uri: str) -> None:
         """Send a resource updated notification."""
         await self.send_notification(
             types.ServerNotification(