Micro.blog Books MCP Server

"""Micro.blog Books API MCP Server using FastMCP.""" import json import logging from typing import Optional from urllib.parse import urljoin import httpx from fastmcp import FastMCP logger = logging.getLogger(__name__) BASE_URL = "https://micro.blog" class MicroBooksClient: """HTTP client for Micro.blog Books API.""" def __init__(self, bearer_token: str) -> None: self.bearer_token = bearer_token self.headers = { "Authorization": f"Bearer {bearer_token}", "User-Agent": "Micro Books MCP Server", "Content-Type": "application/x-www-form-urlencoded", } async def get_bookshelves(self) -> dict: """Get all bookshelves.""" async with httpx.AsyncClient() as client: response = await client.get( urljoin(BASE_URL, "/books/bookshelves"), headers=self.headers, ) response.raise_for_status() return response.json() async def get_bookshelf_books(self, bookshelf_id: int) -> dict: """Get books in a specific bookshelf.""" async with httpx.AsyncClient() as client: response = await client.get( urljoin(BASE_URL, f"/books/bookshelves/{bookshelf_id}"), headers=self.headers, ) response.raise_for_status() return response.json() async def add_bookshelf(self, name: str) -> dict: """Add a new bookshelf.""" async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, "/books/bookshelves"), headers=self.headers, data={"name": name}, ) response.raise_for_status() return {"success": True, "message": f"Bookshelf '{name}' created successfully"} async def rename_bookshelf(self, bookshelf_id: int, name: str) -> dict: """Rename a bookshelf.""" async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, f"/books/bookshelves/{bookshelf_id}"), headers=self.headers, data={"name": name}, ) response.raise_for_status() return {"success": True, "message": f"Bookshelf renamed to '{name}' successfully"} async def add_book( self, title: str, author: str, bookshelf_id: int, isbn: Optional[str] = None, cover_url: Optional[str] = None, ) -> dict: """Add a new book.""" data = { "title": title, "author": author, "bookshelf_id": str(bookshelf_id), } if isbn: data["isbn"] = isbn if cover_url: data["cover_url"] = cover_url async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, "/books"), headers=self.headers, data=data, ) response.raise_for_status() return {"success": True, "message": f"Book '{title}' by {author} added successfully"} async def move_book(self, book_id: int, bookshelf_id: int) -> dict: """Move a book to a different bookshelf.""" async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, f"/books/bookshelves/{bookshelf_id}/assign"), headers=self.headers, data={"book_id": str(book_id)}, ) response.raise_for_status() return {"success": True, "message": f"Book moved to bookshelf {bookshelf_id} successfully"} async def remove_book(self, bookshelf_id: int, book_id: int) -> dict: """Remove a book from a bookshelf.""" async with httpx.AsyncClient() as client: response = await client.delete( urljoin(BASE_URL, f"/books/bookshelves/{bookshelf_id}/remove/{book_id}"), headers=self.headers, ) response.raise_for_status() return {"success": True, "message": "Book removed from bookshelf successfully"} async def change_book_cover(self, bookshelf_id: int, book_id: int, cover_url: str) -> dict: """Change the cover for a book.""" async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, f"/books/bookshelves/{bookshelf_id}/cover/{book_id}"), headers=self.headers, data={"cover_url": cover_url}, ) response.raise_for_status() return {"success": True, "message": "Book cover updated successfully"} async def get_reading_goals(self) -> dict: """Get reading goals.""" async with httpx.AsyncClient() as client: response = await client.get( urljoin(BASE_URL, "/books/goals"), headers=self.headers, ) response.raise_for_status() return response.json() async def get_goal_progress(self, goal_id: int) -> dict: """Get books list progress toward a goal.""" async with httpx.AsyncClient() as client: response = await client.get( urljoin(BASE_URL, f"/books/goals/{goal_id}"), headers=self.headers, ) response.raise_for_status() return response.json() async def update_reading_goal(self, goal_id: int, value: int, progress: Optional[int] = None) -> dict: """Update reading goal.""" data = {"value": str(value)} if progress is not None: data["progress"] = str(progress) async with httpx.AsyncClient() as client: response = await client.post( urljoin(BASE_URL, f"/books/goals/{goal_id}"), headers=self.headers, data=data, ) response.raise_for_status() return {"success": True, "message": "Reading goal updated successfully"} def create_server(bearer_token: str) -> FastMCP: """Create the FastMCP server.""" mcp = FastMCP("Micro Books API") client = MicroBooksClient(bearer_token) @mcp.tool() async def get_bookshelves() -> str: """Get all bookshelves from Micro.blog.""" try: result = await client.get_bookshelves() return json.dumps(result, indent=2) except Exception: logger.exception("Failed to get bookshelves") raise @mcp.tool() async def get_bookshelf_books(bookshelf_id: int) -> str: """Get books in a specific bookshelf. Args: bookshelf_id: The ID of the bookshelf to get books from """ try: result = await client.get_bookshelf_books(bookshelf_id) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to get bookshelf books") raise @mcp.tool() async def add_bookshelf(name: str) -> str: """Add a new bookshelf. Args: name: The name of the new bookshelf """ try: result = await client.add_bookshelf(name) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to add bookshelf") raise @mcp.tool() async def rename_bookshelf(bookshelf_id: int, name: str) -> str: """Rename a bookshelf. Args: bookshelf_id: The ID of the bookshelf to rename name: The new name for the bookshelf """ try: result = await client.rename_bookshelf(bookshelf_id, name) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to rename bookshelf") raise @mcp.tool() async def add_book( title: str, author: str, bookshelf_id: int, isbn: Optional[str] = None, cover_url: Optional[str] = None, ) -> str: """Add a new book. Args: title: The title of the book author: The author of the book bookshelf_id: The ID of the bookshelf to add the book to isbn: The ISBN of the book (optional) cover_url: URL to the book cover image (optional) """ try: result = await client.add_book(title, author, bookshelf_id, isbn, cover_url) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to add book") raise @mcp.tool() async def move_book(book_id: int, bookshelf_id: int) -> str: """Move a book to a different bookshelf. Args: book_id: The ID of the book to move bookshelf_id: The ID of the target bookshelf """ try: result = await client.move_book(book_id, bookshelf_id) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to move book") raise @mcp.tool() async def remove_book(bookshelf_id: int, book_id: int) -> str: """Remove a book from a bookshelf. Args: bookshelf_id: The ID of the bookshelf book_id: The ID of the book to remove """ try: result = await client.remove_book(bookshelf_id, book_id) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to remove book") raise @mcp.tool() async def change_book_cover(bookshelf_id: int, book_id: int, cover_url: str) -> str: """Change the cover for a book. Args: bookshelf_id: The ID of the bookshelf book_id: The ID of the book cover_url: URL to the new cover image """ try: result = await client.change_book_cover(bookshelf_id, book_id, cover_url) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to change book cover") raise @mcp.tool() async def get_reading_goals() -> str: """Get reading goals.""" try: result = await client.get_reading_goals() return json.dumps(result, indent=2) except Exception: logger.exception("Failed to get reading goals") raise @mcp.tool() async def get_goal_progress(goal_id: int) -> str: """Get books list progress toward a goal. Args: goal_id: The ID of the reading goal """ try: result = await client.get_goal_progress(goal_id) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to get goal progress") raise @mcp.tool() async def update_reading_goal(goal_id: int, value: int, progress: Optional[int] = None) -> str: """Update reading goal. Args: goal_id: The ID of the reading goal value: The target number of books for the goal progress: The current progress (number of books read, optional) """ try: result = await client.update_reading_goal(goal_id, value, progress) return json.dumps(result, indent=2) except Exception: logger.exception("Failed to update reading goal") raise return mcp