Migrate to MCP Python SDK v2

.pre-commit-config.yaml

@@ -30,7 +30,7 @@ repos:
       - id: mypy
         args: []
         additional_dependencies:
-          - "mcp>=1.9"
+          - "mcp>=2"
           - "youtube-transcript-api>=1.1.0"
           - "beautifulsoup4>=4.13.3"
           - "humanize>=4.13"

pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
 dependencies = [
     "beautifulsoup4>=4.13.3",
     "humanize>=4.13",
-    "mcp>=1.9",
+    "mcp>=2",
     "pydantic>=2.10.6",
     "requests>=2.32.3",
     "rich-click>=1.8.8",

src/mcp_youtube_transcript/__init__.py

@@ -20,8 +20,7 @@
 import requests
 from bs4 import BeautifulSoup
 from mcp import ServerSession
-from mcp.server import FastMCP
-from mcp.server.fastmcp import Context
+from mcp.server.mcpserver import MCPServer, Context
 from pydantic import Field, BaseModel, AwareDatetime
 from youtube_transcript_api import YouTubeTranscriptApi, FetchedTranscriptSnippet
 from youtube_transcript_api.proxies import WebshareProxyConfig, GenericProxyConfig, ProxyConfig
@@ -37,7 +36,7 @@ class AppContext:
 
 
 @asynccontextmanager
-async def _app_lifespan(_server: FastMCP, proxy_config: ProxyConfig | None) -> AsyncIterator[AppContext]:
+async def _app_lifespan(_server: MCPServer, proxy_config: ProxyConfig | None) -> AsyncIterator[AppContext]:
     # Prepare YoutubeDL params with proxy support
     ytdlp_params: dict[str, Any] = {"quiet": True}
     ytdlp_params.update(_proxy_config_to_ytdlp_params(proxy_config))
@@ -180,7 +179,7 @@ def server(
     webshare_proxy_password: str | None = None,
     http_proxy: str | None = None,
     https_proxy: str | None = None,
-) -> FastMCP:
+) -> MCPServer:
     """Initializes the MCP server."""
 
     proxy_config: ProxyConfig | None = None
@@ -189,7 +188,7 @@ def server(
     elif http_proxy or https_proxy:
         proxy_config = GenericProxyConfig(http_proxy, https_proxy)
 
-    mcp = FastMCP("Youtube Transcript", lifespan=partial(_app_lifespan, proxy_config=proxy_config))
+    mcp = MCPServer("Youtube Transcript", lifespan=partial(_app_lifespan, proxy_config=proxy_config))
 
     @mcp.tool()
     async def get_transcript(

tests/test_mcp.py

@@ -74,7 +74,7 @@ async def test_get_transcript(mcp_client_session: ClientSession) -> None:
 
     transcript = Transcript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -97,7 +97,7 @@ async def test_get_transcript_with_language(mcp_client_session: ClientSession) -
 
     transcript = Transcript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -125,15 +125,15 @@ async def test_get_transcript_fallback_language(
 
     transcript = Transcript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.anyio
 async def test_get_transcript_invalid_url(mcp_client_session: ClientSession) -> None:
     res = await mcp_client_session.call_tool(
         "get_transcript", arguments={"url": "https://www.youtube.com/watch?vv=abcdefg"}
     )
-    assert res.isError
+    assert res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -142,7 +142,7 @@ async def test_get_transcript_invalid_url(mcp_client_session: ClientSession) ->
 @pytest.mark.anyio
 async def test_get_transcript_not_found(mcp_client_session: ClientSession) -> None:
     res = await mcp_client_session.call_tool("get_transcript", arguments={"url": "https://www.youtube.com/watch?v=a"})
-    assert res.isError
+    assert res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -165,7 +165,7 @@ async def test_get_transcript_with_short_url(mcp_client_session: ClientSession)
 
     transcript = Transcript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -187,7 +187,7 @@ async def test_get_transcript_with_response_limit(mcp_client_session_with_respon
             "get_transcript",
             arguments={"url": f"https://www.youtube.com/watch?v={video_id}", "next_cursor": cursor},
         )
-        assert not res.isError
+        assert not res.is_error
         assert isinstance(res.content[0], TextContent)
 
         t = Transcript.model_validate_json(res.content[0].text)
@@ -220,7 +220,7 @@ async def test_get_timed_transcript(mcp_client_session: ClientSession) -> None:
 
     transcript = TimedTranscript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -246,7 +246,7 @@ async def test_get_timed_transcript_with_language(mcp_client_session: ClientSess
 
     transcript = TimedTranscript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -274,15 +274,15 @@ async def test_get_timed_transcript_fallback_language(
 
     transcript = TimedTranscript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.anyio
 async def test_get_timed_transcript_invalid_url(mcp_client_session: ClientSession) -> None:
     res = await mcp_client_session.call_tool(
         "get_timed_transcript", arguments={"url": "https://www.youtube.com/watch?vv=abcdefg"}
     )
-    assert res.isError
+    assert res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -293,7 +293,7 @@ async def test_get_timed_transcript_not_found(mcp_client_session: ClientSession)
     res = await mcp_client_session.call_tool(
         "get_timed_transcript", arguments={"url": "https://www.youtube.com/watch?v=a"}
     )
-    assert res.isError
+    assert res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -316,7 +316,7 @@ async def test_get_timed_transcript_with_short_url(mcp_client_session: ClientSes
 
     transcript = TimedTranscript.model_validate_json(res.content[0].text)
     assert transcript == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -338,7 +338,7 @@ async def test_get_timed_transcript_with_response_limit(mcp_client_session_with_
             "get_timed_transcript",
             arguments={"url": f"https://www.youtube.com/watch?v={video_id}", "next_cursor": cursor},
         )
-        assert not res.isError
+        assert not res.is_error
         assert isinstance(res.content[0], TextContent)
 
         t = TimedTranscript.model_validate_json(res.content[0].text)
@@ -378,7 +378,7 @@ async def test_get_video_info(mcp_client_session: ClientSession) -> None:
 
     info = VideoInfo.model_validate_json(res.content[0].text, strict=True)
     assert info == expect
-    assert not res.isError
+    assert not res.is_error
 
 
 @pytest.mark.skipif(os.getenv("CI") == "true", reason="Skipping this test on CI")
@@ -394,7 +394,7 @@ async def test_get_available_languages(mcp_client_session: ClientSession) -> Non
         "get_available_languages",
         arguments={"url": f"https://www.youtube.com/watch?v={video_id}"},
     )
-    assert not res.isError
+    assert not res.is_error
 
     langs = [r.text for r in res.content if isinstance(r, TextContent)]
     assert langs == expect