server.py•12.5 kB
#!/usr/bin/env python3
"""
Futurama Quote Machine MCP Server
This server provides MCP tools for interacting with the Futurama Quote Machine API.
It enables Claude Desktop to search, create, update, and delete Futurama quotes.
"""
import asyncio
import json
import logging
from typing import Any, Dict, List, Optional
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
CallToolRequest,
CallToolResult,
ListToolsRequest,
TextContent,
Tool,
)
# Configuration
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# API Configuration - Change this to your Futurama Quote Machine API URL
API_BASE_URL = "https://fqm.jeremyschroeder.net/api"
TIMEOUT_SECONDS = 30
# Initialize the MCP server
app = Server("futurama-quote-machine")
async def make_api_request(
method: str,
endpoint: str,
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Make an HTTP request to the Futurama Quote Machine API."""
url = f"{API_BASE_URL}{endpoint}"
async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS) as client:
try:
response = await client.request(
method=method,
url=url,
params=params,
json=json_data,
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error {e.response.status_code}: {e.response.text}")
raise Exception(f"API request failed: {e.response.status_code} - {e.response.text}")
except httpx.RequestError as e:
logger.error(f"Request error: {e}")
raise Exception(f"Request failed: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise Exception(f"Unexpected error: {str(e)}")
@app.list_tools()
async def list_tools() -> List[Tool]:
"""List available tools for the Futurama Quote Machine."""
return [
Tool(
name="get_quotes",
description="Get quotes with optional filtering and pagination",
inputSchema={
"type": "object",
"properties": {
"page": {
"type": "integer",
"description": "Page number for pagination (default: 1)",
"minimum": 1,
},
"per_page": {
"type": "integer",
"description": "Number of quotes per page (default: 20, max: 100)",
"minimum": 1,
"maximum": 100,
},
"character": {
"type": "string",
"description": "Filter by character name (case-insensitive partial match)",
},
"search": {
"type": "string",
"description": "Search in quote text (case-insensitive partial match)",
},
},
},
),
Tool(
name="get_random_quote",
description="Get a random Futurama quote",
inputSchema={
"type": "object",
"properties": {},
},
),
Tool(
name="get_quote_by_id",
description="Get a specific quote by its ID",
inputSchema={
"type": "object",
"properties": {
"quote_id": {
"type": "integer",
"description": "The ID of the quote to retrieve",
"minimum": 1,
},
},
"required": ["quote_id"],
},
),
Tool(
name="create_quote",
description="Create a new Futurama quote",
inputSchema={
"type": "object",
"properties": {
"quote_text": {
"type": "string",
"description": "The text of the quote",
"minLength": 1,
},
"character": {
"type": "string",
"description": "The character who said the quote",
"minLength": 1,
},
},
"required": ["quote_text", "character"],
},
),
Tool(
name="update_quote",
description="Update an existing quote",
inputSchema={
"type": "object",
"properties": {
"quote_id": {
"type": "integer",
"description": "The ID of the quote to update",
"minimum": 1,
},
"quote_text": {
"type": "string",
"description": "The new text of the quote",
"minLength": 1,
},
"character": {
"type": "string",
"description": "The character who said the quote",
"minLength": 1,
},
},
"required": ["quote_id", "quote_text", "character"],
},
),
Tool(
name="delete_quote",
description="Delete a quote by its ID",
inputSchema={
"type": "object",
"properties": {
"quote_id": {
"type": "integer",
"description": "The ID of the quote to delete",
"minimum": 1,
},
},
"required": ["quote_id"],
},
),
Tool(
name="get_characters",
description="Get a list of all characters who have quotes",
inputSchema={
"type": "object",
"properties": {},
},
),
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> List[TextContent]:
"""Handle tool execution requests."""
try:
if name == "get_quotes":
return await handle_get_quotes(arguments)
elif name == "get_random_quote":
return await handle_get_random_quote()
elif name == "get_quote_by_id":
return await handle_get_quote_by_id(arguments)
elif name == "create_quote":
return await handle_create_quote(arguments)
elif name == "update_quote":
return await handle_update_quote(arguments)
elif name == "delete_quote":
return await handle_delete_quote(arguments)
elif name == "get_characters":
return await handle_get_characters()
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error in tool {name}: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def handle_get_quotes(arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle getting quotes with optional filtering and pagination."""
params = {}
if arguments.get("page"):
params["page"] = arguments["page"]
if arguments.get("per_page"):
params["per_page"] = arguments["per_page"]
if arguments.get("character"):
params["character"] = arguments["character"]
if arguments.get("search"):
params["search"] = arguments["search"]
data = await make_api_request("GET", "/quotes", params=params)
result_text = f"Found {data.get('total', 0)} total quotes"
if params.get("character"):
result_text += f" for character '{params['character']}'"
if params.get("search"):
result_text += f" matching '{params['search']}'"
page = data.get("page", 1)
pages = data.get("pages", "unknown")
quotes = data.get("quotes", [])
result_text += f" (page {page} of {pages}, showing {len(quotes)} quotes):\n\n"
for quote in quotes:
quote_id = quote.get("id", "?")
quote_text = quote.get("quote_text", "")
character = quote.get("character", "")
result_text += f"ID {quote_id}: \"{quote_text}\" - {character}\n"
return [TextContent(type="text", text=result_text)]
async def handle_get_random_quote() -> List[TextContent]:
"""Handle getting a random quote."""
data = await make_api_request("GET", "/random")
quote_id = data.get("id", "?")
quote_text = data.get("quote_text", "")
character = data.get("character", "")
result_text = f"Random Quote (ID {quote_id}):\n\"{quote_text}\" - {character}"
return [TextContent(type="text", text=result_text)]
async def handle_get_quote_by_id(arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle getting a specific quote by ID."""
quote_id = arguments.get("quote_id")
if not quote_id:
raise ValueError("quote_id is required")
data = await make_api_request("GET", f"/{quote_id}")
quote_text = data.get("quote_text", "")
character = data.get("character", "")
created_at = data.get("created_at", "")
updated_at = data.get("updated_at", "")
result_text = f"Quote ID {quote_id}:\n\"{quote_text}\" - {character}\n"
if created_at:
result_text += f"Created: {created_at}\n"
if updated_at:
result_text += f"Updated: {updated_at}"
return [TextContent(type="text", text=result_text)]
async def handle_create_quote(arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle creating a new quote."""
quote_text = arguments.get("quote_text")
character = arguments.get("character")
if not quote_text or not character:
raise ValueError("Both quote_text and character are required")
json_data = {
"quote_text": quote_text,
"character": character,
}
data = await make_api_request("POST", "/quotes", json_data=json_data)
quote_id = data.get("id", "?")
result_text = f"Successfully created quote ID {quote_id}:\n\"{quote_text}\" - {character}"
return [TextContent(type="text", text=result_text)]
async def handle_update_quote(arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle updating an existing quote."""
quote_id = arguments.get("quote_id")
quote_text = arguments.get("quote_text")
character = arguments.get("character")
if not quote_id or not quote_text or not character:
raise ValueError("quote_id, quote_text, and character are all required")
json_data = {
"quote_text": quote_text,
"character": character,
}
data = await make_api_request("PUT", f"/{quote_id}", json_data=json_data)
result_text = f"Successfully updated quote ID {quote_id}:\n\"{quote_text}\" - {character}"
return [TextContent(type="text", text=result_text)]
async def handle_delete_quote(arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle deleting a quote."""
quote_id = arguments.get("quote_id")
if not quote_id:
raise ValueError("quote_id is required")
data = await make_api_request("DELETE", f"/{quote_id}")
result_text = f"Successfully deleted quote ID {quote_id}"
if data.get("message"):
result_text = data["message"]
return [TextContent(type="text", text=result_text)]
async def handle_get_characters() -> List[TextContent]:
"""Handle getting all characters."""
data = await make_api_request("GET", "/characters")
characters = data.get("characters", [])
result_text = f"Found {len(characters)} characters with quotes:\n\n"
result_text += "\n".join(f"- {character}" for character in characters)
return [TextContent(type="text", text=result_text)]
async def main():
"""Main entry point for the server."""
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options(),
)
if __name__ == "__main__":
asyncio.run(main())