from typing import Any
import sys
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import asyncio
import json
server = Server("neodb")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
return [
types.Tool(
name="get-user-info",
description="Get current user's basic info",
inputSchema={
"type": "object",
"properties": {}, # No parameters needed
"required": [],
},
),
types.Tool(
name="search-books",
description="Search items in catalog",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query for books",
}
},
"required": ["query"],
},
),
types.Tool(
name="get-book",
description="Get detailed information about a specific book",
inputSchema={
"type": "object",
"properties": {
"book_id": {
"type": "string",
"description": "The ID of the book to retrieve",
},
},
"required": ["book_id"],
},
),
]
async def make_neodb_request(client: httpx.AsyncClient, access_token: str, endpoint: str, api_base: str) -> dict[str, Any] | None:
"""
Make an authenticated request to the NeoDB API
Args:
client (httpx.AsyncClient): The HTTP client to use
access_token (str): The access token obtained from get_access_token
endpoint (str): The API endpoint to call (e.g., '/api/me')
api_base (str): The base URL for the NeoDB API
Returns:
dict: The API response
"""
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json",
}
url = f"{api_base}{endpoint}"
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx status codes)
return None, e.response.status_code
except Exception as e:
# Handle other errors (network issues, timeouts, etc)
return None, 500 # Return 500 for internal errors
def format_book(book: dict) -> str:
"""Format a book into a concise string."""
return (
f"Title: {book.get('title', 'Unknown')}\n"
f"Author: {book.get('author', 'Unknown')}\n"
f"Rating: {book.get('rating', 'N/A')}\n"
f"Description: {book.get('description', 'No description available')}\n"
"---"
)
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Tools can fetch book data and return formatted responses.
"""
# Access the configuration from server instance
api_base = server.config.get("api_base")
access_token = server.config.get("access_token")
if not api_base or not access_token:
return [types.TextContent(type="text", text="Server configuration missing API base URL or access token")]
if name == "get-user-info":
"""
https://neodb.social/developer/#/user/users_api_me
"""
async with httpx.AsyncClient() as client:
user_data, status_code = await make_neodb_request(
client,
access_token,
'/api/me',
api_base
)
if status_code != 200:
error_message = {
401: "Unauthorized",
}.get(status_code, f"Request failed with status code: {status_code}")
return [types.TextContent(type="text", text=error_message)]
if not user_data:
return [types.TextContent(type="text", text="Failed to retrieve user information")]
# Format user information
user_text = (
f"User Information:\n"
f"Username: {user_data.get('username', 'Unknown')}\n"
f"Display Name: {user_data.get('display_name', 'Unknown')}\n"
f"Email: {user_data.get('email', 'Not provided')}\n"
f"URL: {user_data.get('url', 'Not provided')}\n"
f"Account Created: {user_data.get('created_at', 'Unknown')}\n"
)
return [
types.TextContent(
type="text",
text=user_text
)
]
elif name == "search-books":
"""
https://neodb.social/developer/#/catalog/catalog_api_search_item
#TBD category and page parameters not supported
"""
query = arguments.get("query")
if not query:
raise ValueError("Missing query parameter")
async with httpx.AsyncClient() as client:
search_data, status_code = await make_neodb_request(
client,
access_token,
f"/api/catalog/search?query={query}&page=1",
api_base
)
if status_code != 200:
error_message = {
400: "Bad request",
}.get(status_code, f"Request failed with status code: {status_code}")
return [types.TextContent(type="text", text=error_message)]
if not search_data:
return [types.TextContent(type="text", text=f"Failed to search books")]
books = search_data.get("data", [])
if not books:
return [types.TextContent(type="text", text=f"No books found for query: {query}")]
# Format each book into a concise string
formatted_books = [format_book(book) for book in books]
books_text = f"Search results for '{query}':\n\n" + "\n".join(formatted_books)
return [
types.TextContent(
type="text",
text=books_text
)
]
elif name == "get-book":
book_id = arguments.get("book_id")
if not book_id:
raise ValueError("Missing book_id parameter")
async with httpx.AsyncClient() as client:
book_data = await make_neodb_request(
client,
access_token,
f"/api/book/{book_id}",
api_base
)
if not book_data:
return [types.TextContent(type="text", text=f"Failed to retrieve book with ID: {book_id}")]
# Format detailed book information
book_text = format_book(book_data)
return [
types.TextContent(
type="text",
text=book_text
)
]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
# Check if correct number of arguments are provided
if len(sys.argv) != 3:
print("Usage: python script.py <api_base_url> <access_token>")
sys.exit(1)
# Get arguments from command line
api_base = sys.argv[1]
access_token = sys.argv[2]
# Store configuration in server instance
server.config = {
"api_base": api_base,
"access_token": access_token
}
# Run the server using stdin/stdout streams
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="neodb",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())