mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-14 00:43:20 +02:00
203 lines
7.2 KiB
Python
203 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import httpx2
|
|
import media_picker_server
|
|
import pytest
|
|
from media_picker_server import (
|
|
MediaLink,
|
|
_play_via_mcp,
|
|
_saved,
|
|
mcp,
|
|
play_media,
|
|
save_media,
|
|
verify_links,
|
|
youtube_video_id,
|
|
)
|
|
from prefab_ui.app import PROTOCOL_VERSION
|
|
|
|
from fastmcp import Client, FastMCP
|
|
from fastmcp.exceptions import ToolError
|
|
|
|
KNOWN_TITLES = {
|
|
"jkAw87ZIwQA": ("The Mother of all Science Scandals", "BobbyBroccoli"),
|
|
"D1VM6V6wmU0": ("Live Bird Feeder", "HANS Nature"),
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def oembed(monkeypatch: pytest.MonkeyPatch) -> list[str]:
|
|
"""Answer YouTube oEmbed for KNOWN_TITLES and 404 everything else."""
|
|
requested: list[str] = []
|
|
|
|
def handler(request: httpx2.Request) -> httpx2.Response:
|
|
assert request.url.host == "www.youtube.com"
|
|
assert request.url.path == "/oembed"
|
|
video_url = request.url.params["url"]
|
|
requested.append(video_url)
|
|
video_id = parse_qs(urlparse(video_url).query)["v"][0]
|
|
if video_id not in KNOWN_TITLES:
|
|
return httpx2.Response(404)
|
|
title, channel = KNOWN_TITLES[video_id]
|
|
return httpx2.Response(200, json={"title": title, "author_name": channel})
|
|
|
|
monkeypatch.setattr(
|
|
media_picker_server,
|
|
"_oembed_client",
|
|
lambda: httpx2.AsyncClient(transport=httpx2.MockTransport(handler)),
|
|
)
|
|
return requested
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("url", "expected"),
|
|
[
|
|
("https://www.youtube.com/watch?v=jkAw87ZIwQA", "jkAw87ZIwQA"),
|
|
("https://youtube.com/watch?v=jkAw87ZIwQA&t=30s", "jkAw87ZIwQA"),
|
|
("https://m.youtube.com/watch?v=jkAw87ZIwQA", "jkAw87ZIwQA"),
|
|
("https://youtu.be/jkAw87ZIwQA?si=abc", "jkAw87ZIwQA"),
|
|
("https://www.youtube.com/live/D1VM6V6wmU0", "D1VM6V6wmU0"),
|
|
("https://www.youtube.com/shorts/D1VM6V6wmU0", "D1VM6V6wmU0"),
|
|
("https://www.youtube.com/embed/D1VM6V6wmU0", "D1VM6V6wmU0"),
|
|
("https://www.youtube.com/watch?v=short", None),
|
|
("https://www.youtube.com/@BobbyBroccoli", None),
|
|
("https://youtube.com.evil.test/watch?v=jkAw87ZIwQA", None),
|
|
("https://archive.org/details/apollo", None),
|
|
("https://[bad", None),
|
|
],
|
|
)
|
|
def test_youtube_video_id(url: str, expected: str | None) -> None:
|
|
assert youtube_video_id(url) == expected
|
|
|
|
|
|
async def test_links_are_verified_with_youtube(oembed: list[str]) -> None:
|
|
candidates, unsupported, unverified = await verify_links(
|
|
[
|
|
MediaLink(url="https://youtu.be/jkAw87ZIwQA", why="A deep story."),
|
|
MediaLink(url="https://www.youtube.com/live/D1VM6V6wmU0", label="Live"),
|
|
MediaLink(url="https://www.youtube.com/watch?v=jkAw87ZIwQA"),
|
|
MediaLink(url="https://www.youtube.com/watch?v=aaaaaaaaaaa"),
|
|
MediaLink(url="https://archive.org/details/apollo"),
|
|
]
|
|
)
|
|
|
|
assert [(c["source_id"], c["title"], c["channel"]) for c in candidates] == [
|
|
("jkAw87ZIwQA", "The Mother of all Science Scandals", "BobbyBroccoli"),
|
|
("D1VM6V6wmU0", "Live Bird Feeder", "HANS Nature"),
|
|
]
|
|
assert candidates[0]["why"] == "A deep story."
|
|
assert candidates[1]["label"] == "Live"
|
|
assert candidates[0]["url"] == "https://www.youtube.com/watch?v=jkAw87ZIwQA"
|
|
assert (unsupported, unverified) == (1, 1)
|
|
assert len(oembed) == 4
|
|
|
|
|
|
async def test_malformed_link_does_not_discard_valid_candidates(
|
|
oembed: list[str],
|
|
) -> None:
|
|
candidates, unsupported, unverified = await verify_links(
|
|
[
|
|
MediaLink(url="https://[bad"),
|
|
MediaLink(url="https://youtu.be/jkAw87ZIwQA"),
|
|
]
|
|
)
|
|
|
|
assert [candidate["source_id"] for candidate in candidates] == ["jkAw87ZIwQA"]
|
|
assert (unsupported, unverified) == (1, 0)
|
|
assert oembed == ["https://www.youtube.com/watch?v=jkAw87ZIwQA"]
|
|
|
|
|
|
async def test_actions_are_idempotent_and_reject_malformed_ids() -> None:
|
|
_saved.clear()
|
|
|
|
play_receipt = await play_media("youtube", "D1VM6V6wmU0", "Live Bird Feeder")
|
|
assert play_receipt["status"] == "queued"
|
|
assert play_receipt["url"] == "https://www.youtube.com/watch?v=D1VM6V6wmU0"
|
|
|
|
assert save_media("youtube", "D1VM6V6wmU0", "Live Bird Feeder")["status"] == "saved"
|
|
assert (
|
|
save_media("youtube", "D1VM6V6wmU0", "Live Bird Feeder")["status"]
|
|
== "already_saved"
|
|
)
|
|
|
|
with pytest.raises(ToolError, match="not valid"):
|
|
await play_media("youtube", "x; rm -rf /", "Nope")
|
|
|
|
|
|
async def test_playback_routes_verified_item_through_mcp() -> None:
|
|
actuator = FastMCP("Test actuator")
|
|
|
|
@actuator.tool
|
|
def play_media(source: str, source_id: str, url: str, title: str) -> dict[str, str]:
|
|
return {"command": f"play:{source}:{source_id}", "url": url, "title": title}
|
|
|
|
receipt = await _play_via_mcp(
|
|
"youtube",
|
|
"jkAw87ZIwQA",
|
|
"https://www.youtube.com/watch?v=jkAw87ZIwQA",
|
|
"The Mother of all Science Scandals",
|
|
actuator,
|
|
"play_media",
|
|
)
|
|
assert receipt == {
|
|
"command": "play:youtube:jkAw87ZIwQA",
|
|
"url": "https://www.youtube.com/watch?v=jkAw87ZIwQA",
|
|
"title": "The Mother of all Science Scandals",
|
|
}
|
|
|
|
|
|
async def test_actuator_errors_surface_as_a_playback_failure() -> None:
|
|
actuator = FastMCP("Test actuator")
|
|
|
|
@actuator.tool
|
|
def play_media(source: str, source_id: str, url: str, title: str) -> None:
|
|
raise ToolError("The TV is not reachable")
|
|
|
|
with pytest.raises(ToolError, match="could not start this item"):
|
|
await _play_via_mcp(
|
|
"youtube",
|
|
"jkAw87ZIwQA",
|
|
"https://www.youtube.com/watch?v=jkAw87ZIwQA",
|
|
"The Mother of all Science Scandals",
|
|
actuator,
|
|
"play_media",
|
|
)
|
|
|
|
|
|
async def test_mcp_host_loop_exposes_ui_and_marks_backend_tools_app_only(
|
|
oembed: list[str],
|
|
) -> None:
|
|
async with Client(mcp) as client:
|
|
tools_by_name = {tool.name: tool for tool in await client.list_tools()}
|
|
assert set(tools_by_name) == {
|
|
"play_media",
|
|
"save_media",
|
|
"show_media_picker",
|
|
}
|
|
assert tools_by_name["play_media"].meta["ui"]["visibility"] == ["app"]
|
|
assert tools_by_name["save_media"].meta["ui"]["visibility"] == ["app"]
|
|
|
|
picker = tools_by_name["show_media_picker"]
|
|
assert picker.meta is not None
|
|
assert picker.meta["ui"]["resourceUri"].startswith("ui://prefab/tool/")
|
|
link_schema = picker.input_schema["properties"]["links"]["items"]
|
|
assert link_schema["required"] == ["url"]
|
|
|
|
result = await client.call_tool(
|
|
"show_media_picker",
|
|
{
|
|
"links": [
|
|
{"url": "https://youtu.be/jkAw87ZIwQA"},
|
|
{"url": "https://www.youtube.com/watch?v=aaaaaaaaaaa"},
|
|
{"url": "https://vimeo.com/76979871"},
|
|
]
|
|
},
|
|
)
|
|
|
|
assert result.structured_content is not None
|
|
assert result.structured_content["$prefab"]["version"] == PROTOCOL_VERSION
|
|
state = result.structured_content["state"]
|
|
assert state["source_ids"] == ["jkAw87ZIwQA"]
|
|
assert state["unverified_count"] == 1
|
|
assert state["unsupported_count"] == 1
|