802 lines
26 KiB
Python
802 lines
26 KiB
Python
"""FastMCP HTTP server exposing all tools from https://tryme.tpk.pw/openapi.json."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any, Literal
|
|
|
|
import yaml
|
|
from fastmcp import FastMCP
|
|
from pydantic import Field
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse, PlainTextResponse, Response
|
|
|
|
from app.client import TryMeClient
|
|
from app.config import PROJECT_ROOT, settings
|
|
from app.models import QuestionIn
|
|
|
|
_client = TryMeClient(config=settings)
|
|
|
|
|
|
def get_client() -> TryMeClient:
|
|
"""Return the active TryMe API client instance."""
|
|
return _client
|
|
|
|
|
|
def set_client(client: TryMeClient) -> None:
|
|
"""Override the active TryMe API client instance (primarily for testing)."""
|
|
global _client
|
|
_client = client
|
|
|
|
|
|
mcp = FastMCP(
|
|
name="TryMe TPK MCP Server",
|
|
version=settings.version,
|
|
instructions=(
|
|
"MCP HTTP server for the TryMe Gaming App API (https://tryme.tpk.pw). "
|
|
"Provides tools for student quiz creation, discovery, gameplay/submission, "
|
|
"voting, favourites, leaderboards/scores, Auth0/dev authentication, personal "
|
|
"API keys, and admin moderation. Authenticated endpoints accept optional "
|
|
"`api_key`, `authorization` (Bearer token), or `tryme_session` parameters, "
|
|
"and also inherit incoming HTTP headers or server environment variables "
|
|
"(`TRYME_API_KEY`, `TRYME_BEARER_TOKEN`, `TRYME_SESSION_COOKIE`)."
|
|
),
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Custom HTTP Endpoints (Healthcheck & OpenAPI Specification)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
OPENAPI_PATH = PROJECT_ROOT / "openapi.yaml"
|
|
|
|
|
|
@mcp.custom_route("/health", methods=["GET"], name="server_health")
|
|
async def server_health(_request: Request) -> Response:
|
|
"""Container and service healthcheck endpoint."""
|
|
return JSONResponse(
|
|
{
|
|
"status": "ok",
|
|
"service": "trymemcp",
|
|
"version": settings.version,
|
|
"transport": "http",
|
|
"host": settings.mcp_host,
|
|
"port": settings.mcp_port,
|
|
"mcp_endpoint": settings.mcp_path,
|
|
"upstream_url": settings.tryme_base_url,
|
|
}
|
|
)
|
|
|
|
|
|
@mcp.custom_route("/openapi.yaml", methods=["GET"], name="openapi_yaml")
|
|
async def serve_openapi_yaml(_request: Request) -> Response:
|
|
"""Serve the server's OpenAPI 3.1 YAML specification."""
|
|
if not OPENAPI_PATH.exists():
|
|
return JSONResponse({"detail": "openapi.yaml not found"}, status_code=404)
|
|
content = OPENAPI_PATH.read_text(encoding="utf-8")
|
|
return PlainTextResponse(content, media_type="application/yaml")
|
|
|
|
|
|
@mcp.custom_route("/openapi.json", methods=["GET"], name="openapi_json")
|
|
async def serve_openapi_json(_request: Request) -> Response:
|
|
"""Serve the server's OpenAPI 3.1 JSON specification."""
|
|
if not OPENAPI_PATH.exists():
|
|
return JSONResponse({"detail": "openapi.yaml not found"}, status_code=404)
|
|
spec = yaml.safe_load(OPENAPI_PATH.read_text(encoding="utf-8"))
|
|
return JSONResponse(spec)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared Auth Parameter Type Aliases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ApiKeyParam = Annotated[
|
|
str | None,
|
|
Field(
|
|
default=None,
|
|
description="Optional TryMe API key (`X-API-Key` header). Overrides server default if provided.",
|
|
),
|
|
]
|
|
AuthorizationParam = Annotated[
|
|
str | None,
|
|
Field(
|
|
default=None,
|
|
description="Optional Bearer token or `Authorization` header value.",
|
|
),
|
|
]
|
|
SessionCookieParam = Annotated[
|
|
str | None,
|
|
Field(
|
|
default=None,
|
|
description="Optional `tryme_session` cookie value.",
|
|
),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# System Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"System"})
|
|
async def healthcheck() -> dict[str, Any]:
|
|
"""Check the health status of the upstream TryMe Gaming App API (`GET /health`)."""
|
|
return await get_client().request("GET", "/health")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Authentication Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"Authentication"})
|
|
async def get_auth_config() -> dict[str, Any]:
|
|
"""Retrieve Auth0 federated login settings, dev-auth status, and admin claim status (`GET /api/v1/auth/config`)."""
|
|
return await get_client().request("GET", "/api/v1/auth/config")
|
|
|
|
|
|
@mcp.tool(tags={"Authentication"})
|
|
async def auth0_login(
|
|
return_to: Annotated[
|
|
str,
|
|
Field(default="/", description="Relative path to redirect to after login."),
|
|
] = "/",
|
|
) -> dict[str, Any]:
|
|
"""Initiate the Auth0 federated login flow (`GET /api/v1/auth/login`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/auth/login",
|
|
params={"return_to": return_to},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Authentication"})
|
|
async def auth0_callback(
|
|
code: Annotated[str, Field(description="Authorization code returned by Auth0.")],
|
|
state: Annotated[str, Field(description="CSRF state parameter returned by Auth0.")],
|
|
) -> dict[str, Any]:
|
|
"""Handle the Auth0 OAuth callback (`GET /api/v1/auth/callback`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/auth/callback",
|
|
params={"code": code, "state": state},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Authentication"})
|
|
async def dev_login(
|
|
nickname: Annotated[
|
|
str,
|
|
Field(min_length=2, max_length=48, description="User nickname (2-48 chars)."),
|
|
],
|
|
email: Annotated[
|
|
str | None,
|
|
Field(default=None, max_length=255, description="Optional email address."),
|
|
] = None,
|
|
as_moderator: Annotated[
|
|
bool,
|
|
Field(default=False, description="Request moderator privileges in dev mode."),
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""Authenticate using the development login endpoint (`POST /api/v1/auth/dev-login`).
|
|
|
|
Returns an `access_token`, `token_type`, and `user` profile when dev auth is enabled.
|
|
"""
|
|
payload: dict[str, Any] = {
|
|
"nickname": nickname,
|
|
"email": email,
|
|
"as_moderator": as_moderator,
|
|
}
|
|
return await get_client().request("POST", "/api/v1/auth/dev-login", json_body=payload)
|
|
|
|
|
|
@mcp.tool(tags={"Authentication"})
|
|
async def logout(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Log out the current authenticated session (`POST /api/v1/auth/logout`)."""
|
|
await get_client().request(
|
|
"POST",
|
|
"/api/v1/auth/logout",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
return {"status": "logged_out"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Users & API Keys Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def get_me(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Retrieve the authenticated user's profile (`GET /api/v1/users/me`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/users/me",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def update_me(
|
|
nickname: Annotated[
|
|
str | None,
|
|
Field(default=None, min_length=2, max_length=48, description="Updated nickname."),
|
|
] = None,
|
|
first_name: Annotated[
|
|
str | None,
|
|
Field(default=None, max_length=80, description="Updated first name."),
|
|
] = None,
|
|
last_name: Annotated[
|
|
str | None,
|
|
Field(default=None, max_length=80, description="Updated last name."),
|
|
] = None,
|
|
age: Annotated[
|
|
int | None,
|
|
Field(default=None, ge=4, le=120, description="Updated age (4-120)."),
|
|
] = None,
|
|
clear_age: Annotated[
|
|
bool,
|
|
Field(default=False, description="Set to true to clear the stored age."),
|
|
] = False,
|
|
gender: Annotated[
|
|
str | None,
|
|
Field(default=None, description="Updated gender value."),
|
|
] = None,
|
|
gender_custom: Annotated[
|
|
str | None,
|
|
Field(default=None, max_length=80, description="Custom gender text."),
|
|
] = None,
|
|
avatar_animal: Annotated[
|
|
str | None,
|
|
Field(default=None, description="Avatar animal identifier."),
|
|
] = None,
|
|
avatar_color: Annotated[
|
|
str | None,
|
|
Field(default=None, description="Avatar color identifier."),
|
|
] = None,
|
|
public_website_consent: Annotated[
|
|
bool | None,
|
|
Field(default=None, description="Public website display consent flag."),
|
|
] = None,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Update the authenticated user's profile (`PATCH /api/v1/users/me`)."""
|
|
payload: dict[str, Any] = {}
|
|
optional_fields = {
|
|
"nickname": nickname,
|
|
"first_name": first_name,
|
|
"last_name": last_name,
|
|
"age": age,
|
|
"gender": gender,
|
|
"gender_custom": gender_custom,
|
|
"avatar_animal": avatar_animal,
|
|
"avatar_color": avatar_color,
|
|
"public_website_consent": public_website_consent,
|
|
}
|
|
for key, value in optional_fields.items():
|
|
if value is not None:
|
|
payload[key] = value
|
|
if clear_age:
|
|
payload["clear_age"] = True
|
|
|
|
return await get_client().request(
|
|
"PATCH",
|
|
"/api/v1/users/me",
|
|
json_body=payload,
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def get_my_scores(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Retrieve quiz score summaries and attempt counts for the authenticated user (`GET /api/v1/users/me/scores`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/users/me/scores",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def list_api_keys(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""List personal API keys belonging to the authenticated user (`GET /api/v1/users/me/api-keys`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/users/me/api-keys",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def create_api_key(
|
|
name: Annotated[
|
|
str,
|
|
Field(min_length=1, max_length=80, description="Descriptive name for the API key."),
|
|
],
|
|
expires_in_days: Annotated[
|
|
int | None,
|
|
Field(
|
|
default=90,
|
|
ge=1,
|
|
le=3650,
|
|
description="Number of days until expiration (1-3650), or null for non-expiring if supported.",
|
|
),
|
|
] = 90,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new personal API key (`POST /api/v1/users/me/api-keys`). Returns `raw_key` once."""
|
|
payload = {"name": name, "expires_in_days": expires_in_days}
|
|
return await get_client().request(
|
|
"POST",
|
|
"/api/v1/users/me/api-keys",
|
|
json_body=payload,
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Users & API Keys"})
|
|
async def expire_api_key(
|
|
key_id: Annotated[int, Field(description="Numeric ID of the API key to expire/revoke.")],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Expire/revoke an existing personal API key (`POST /api/v1/users/me/api-keys/{key_id}/expire`)."""
|
|
return await get_client().request(
|
|
"POST",
|
|
f"/api/v1/users/me/api-keys/{key_id}/expire",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quizzes Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def list_quizzes(
|
|
q: Annotated[
|
|
str | None,
|
|
Field(default=None, max_length=120, description="Optional search query string."),
|
|
] = None,
|
|
mode: Annotated[
|
|
str | None,
|
|
Field(
|
|
default=None,
|
|
description="Optional quiz mode filter (e.g. 'multiple_choice', 'write_in', 'mixed').",
|
|
),
|
|
] = None,
|
|
mine_only: Annotated[
|
|
bool,
|
|
Field(default=False, description="If true, return only quizzes authored by the authenticated user."),
|
|
] = False,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Search and list quizzes (`GET /api/v1/quizzes`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/quizzes",
|
|
params={"q": q, "mode": mode, "mine_only": mine_only},
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def create_quiz(
|
|
title: Annotated[
|
|
str,
|
|
Field(min_length=1, max_length=140, description="Quiz title (1-140 chars)."),
|
|
],
|
|
description: Annotated[
|
|
str,
|
|
Field(default="", max_length=1000, description="Quiz description (up to 1000 chars)."),
|
|
] = "",
|
|
is_public: Annotated[
|
|
bool,
|
|
Field(default=True, description="Whether the quiz is publicly listed."),
|
|
] = True,
|
|
questions: Annotated[
|
|
list[QuestionIn] | None,
|
|
Field(default=None, max_length=40, description="Up to 40 quiz questions."),
|
|
] = None,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new quiz (`POST /api/v1/quizzes`)."""
|
|
payload: dict[str, Any] = {
|
|
"title": title,
|
|
"description": description,
|
|
"is_public": is_public,
|
|
"questions": [q.model_dump() for q in (questions or [])],
|
|
}
|
|
return await get_client().request(
|
|
"POST",
|
|
"/api/v1/quizzes",
|
|
json_body=payload,
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def get_quiz(
|
|
slug_or_id: Annotated[
|
|
str,
|
|
Field(description="Quiz share slug or numeric quiz ID (passed as string)."),
|
|
],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Retrieve full details and questions for a quiz (`GET /api/v1/quizzes/{slug_or_id}`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
f"/api/v1/quizzes/{slug_or_id}",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def update_quiz(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to update.")],
|
|
title: Annotated[
|
|
str,
|
|
Field(min_length=1, max_length=140, description="Updated quiz title (1-140 chars)."),
|
|
],
|
|
description: Annotated[
|
|
str,
|
|
Field(default="", max_length=1000, description="Updated quiz description (up to 1000 chars)."),
|
|
] = "",
|
|
is_public: Annotated[
|
|
bool,
|
|
Field(default=True, description="Whether the quiz is publicly listed."),
|
|
] = True,
|
|
questions: Annotated[
|
|
list[QuestionIn] | None,
|
|
Field(default=None, max_length=40, description="Full updated list of up to 40 quiz questions."),
|
|
] = None,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Update an existing quiz (`PUT /api/v1/quizzes/{quiz_id}`)."""
|
|
payload: dict[str, Any] = {
|
|
"title": title,
|
|
"description": description,
|
|
"is_public": is_public,
|
|
}
|
|
if questions is not None:
|
|
payload["questions"] = [q.model_dump() for q in questions]
|
|
return await get_client().request(
|
|
"PUT",
|
|
f"/api/v1/quizzes/{quiz_id}",
|
|
json_body=payload,
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def delete_quiz(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to delete.")],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Delete a quiz authored by the authenticated user (`DELETE /api/v1/quizzes/{quiz_id}`)."""
|
|
await get_client().request(
|
|
"DELETE",
|
|
f"/api/v1/quizzes/{quiz_id}",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
return {"status": "deleted", "quiz_id": quiz_id}
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def submit_quiz(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to submit answers for.")],
|
|
answers: Annotated[
|
|
dict[str, str],
|
|
Field(
|
|
default_factory=dict,
|
|
description="Mapping of question ID (as string) to submitted option ID or write-in answer.",
|
|
),
|
|
],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Submit answers for a quiz and receive graded results & scores (`POST /api/v1/quizzes/{quiz_id}/submit`)."""
|
|
return await get_client().request(
|
|
"POST",
|
|
f"/api/v1/quizzes/{quiz_id}/submit",
|
|
json_body={"answers": answers},
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Quizzes"})
|
|
async def vote_quiz(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to vote on.")],
|
|
value: Annotated[
|
|
Literal[-1, 0, 1],
|
|
Field(description="Vote value: `1` (upvote), `-1` (downvote), or `0` (remove vote)."),
|
|
],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Upvote, downvote, or clear vote on a quiz (`POST /api/v1/quizzes/{quiz_id}/vote`)."""
|
|
return await get_client().request(
|
|
"POST",
|
|
f"/api/v1/quizzes/{quiz_id}/vote",
|
|
json_body={"value": value},
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Favourites Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"Favourites"})
|
|
async def list_favourites(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""List the authenticated user's favourited quizzes (`GET /api/v1/favourites`)."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/favourites",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Favourites"})
|
|
async def add_favourite(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to favourite.")],
|
|
position: Annotated[
|
|
int | None,
|
|
Field(default=None, ge=0, description="Optional 0-based position in the favourites list."),
|
|
] = None,
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Add a quiz to the authenticated user's favourites (`POST /api/v1/favourites`)."""
|
|
payload: dict[str, Any] = {"quiz_id": quiz_id}
|
|
if position is not None:
|
|
payload["position"] = position
|
|
return await get_client().request(
|
|
"POST",
|
|
"/api/v1/favourites",
|
|
json_body=payload,
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Favourites"})
|
|
async def reorder_favourites(
|
|
quiz_ids: Annotated[
|
|
list[int],
|
|
Field(description="Ordered list of favourited quiz IDs representing the desired order."),
|
|
],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Reorder the authenticated user's favourited quizzes (`PUT /api/v1/favourites/reorder`)."""
|
|
return await get_client().request(
|
|
"PUT",
|
|
"/api/v1/favourites/reorder",
|
|
json_body={"quiz_ids": quiz_ids},
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Favourites"})
|
|
async def remove_favourite(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to remove from favourites.")],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Remove a quiz from the authenticated user's favourites (`DELETE /api/v1/favourites/{quiz_id}`)."""
|
|
return await get_client().request(
|
|
"DELETE",
|
|
f"/api/v1/favourites/{quiz_id}",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Moderation Tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def claim_initial_moderator(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Claim initial moderator/admin status when no admin exists (`POST /api/v1/admin/claim`)."""
|
|
return await get_client().request(
|
|
"POST",
|
|
"/api/v1/admin/claim",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def list_users_for_moderation(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""List all users for admin moderation (`GET /api/v1/admin/users`). Requires moderator privileges."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/admin/users",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def ban_user(
|
|
user_id: Annotated[int, Field(description="Numeric ID of the user to ban.")],
|
|
reason: Annotated[
|
|
str,
|
|
Field(
|
|
default="Community guidelines violation",
|
|
max_length=255,
|
|
description="Reason for banning the user.",
|
|
),
|
|
] = "Community guidelines violation",
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Ban a user (`POST /api/v1/admin/users/{user_id}/ban`). Requires moderator privileges."""
|
|
return await get_client().request(
|
|
"POST",
|
|
f"/api/v1/admin/users/{user_id}/ban",
|
|
json_body={"reason": reason},
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def unban_user(
|
|
user_id: Annotated[int, Field(description="Numeric ID of the user to unban.")],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Unban a previously banned user (`POST /api/v1/admin/users/{user_id}/unban`). Requires moderator privileges."""
|
|
return await get_client().request(
|
|
"POST",
|
|
f"/api/v1/admin/users/{user_id}/unban",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def list_quizzes_for_moderation(
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""List all quizzes for admin moderation (`GET /api/v1/admin/quizzes`). Requires moderator privileges."""
|
|
return await get_client().request(
|
|
"GET",
|
|
"/api/v1/admin/quizzes",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
|
|
|
|
@mcp.tool(tags={"Moderation"})
|
|
async def moderator_remove_quiz(
|
|
quiz_id: Annotated[int, Field(description="Numeric ID of the quiz to remove.")],
|
|
api_key: ApiKeyParam = None,
|
|
authorization: AuthorizationParam = None,
|
|
tryme_session: SessionCookieParam = None,
|
|
) -> dict[str, Any]:
|
|
"""Remove any quiz as a moderator (`DELETE /api/v1/admin/quizzes/{quiz_id}`). Requires moderator privileges."""
|
|
await get_client().request(
|
|
"DELETE",
|
|
f"/api/v1/admin/quizzes/{quiz_id}",
|
|
api_key=api_key,
|
|
authorization=authorization,
|
|
tryme_session=tryme_session,
|
|
)
|
|
return {"status": "removed", "quiz_id": quiz_id}
|
|
|
|
|
|
# ASGI application for Uvicorn / Gunicorn (`uvicorn app.main:app --host 0.0.0.0 --port 8000`)
|
|
app = mcp.http_app(path=settings.mcp_path, transport="http")
|
|
|
|
|
|
def run_server() -> None:
|
|
"""Start the FastMCP HTTP server bound to 0.0.0.0 (all interfaces) by default."""
|
|
mcp.run(
|
|
transport="http",
|
|
host=settings.mcp_host,
|
|
port=settings.mcp_port,
|
|
path=settings.mcp_path,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_server()
|