MCPeasy

""" Custom quotes resource with auto-seeding from remote CSV """ from typing import Dict, Any, Optional, List from src.resources.base import BaseResource from src.resources.types import MCPResource, ResourceContent import logging logger = logging.getLogger(__name__) class QuotesResource(BaseResource): """ Inspirational quotes resource that auto-seeds from a remote CSV file Provides access to a collection of quotes with search and filtering capabilities. Auto-seeds from GitHub gist with quotes when table is empty. """ name = "myorg/quotes" description = "Collection of inspirational quotes from famous authors" uri_scheme = "quotes" # Auto-seed from remote CSV when table is empty seed_source = "https://gist.githubusercontent.com/JakubPetriska/060958fd744ca34f099e947cd080b540/raw/963b5a9355f04741239407320ac973a6096cd7b6/quotes.csv" @classmethod def get_config_schema(cls) -> Optional[Dict[str, Any]]: """Configuration schema for quotes resource""" return { "type": "object", "properties": { "max_quotes": { "type": "integer", "description": "Maximum number of quotes to return", "default": 50, "minimum": 1, "maximum": 1000 }, "allowed_authors": { "type": "array", "items": {"type": "string"}, "description": "List of authors this client can access (empty = all authors)" }, "search_enabled": { "type": "boolean", "description": "Allow searching quotes by content", "default": True } } } async def _get_model_class(self): """Return the Quote model for seeding""" from .models import Quote return Quote async def list_resources(self, config: Optional[Dict[str, Any]] = None) -> List[MCPResource]: """List available quote resources with client-specific filtering""" try: resources = [] # Get configuration max_quotes = 50 allowed_authors = [] search_enabled = True if config: max_quotes = config.get("max_quotes", 50) allowed_authors = config.get("allowed_authors", []) search_enabled = config.get("search_enabled", True) # Get quotes from database if not self._db: logger.warning("No database connection available") return [] from .models import Quote async with self._db.get_session() as session: from sqlalchemy import select # Build query with author filtering query = select(Quote) if allowed_authors: query = query.where(Quote.author.in_(allowed_authors)) # Limit results query = query.limit(max_quotes) result = await session.execute(query) quotes = result.scalars().all() # Create resource entries for quote in quotes: resources.append(MCPResource( uri=f"quotes://{quote.id}", name=f"Quote by {quote.author}", description=f"{quote.quote[:100]}{'...' if len(quote.quote) > 100 else ''}", mimeType="text/plain" )) # Add search resource if enabled if search_enabled: resources.append(MCPResource( uri="quotes://search", name="Search Quotes", description="Search quotes by content or author", mimeType="application/json" )) # Add authors list resource resources.append(MCPResource( uri="quotes://authors", name="Authors List", description="List of all available authors", mimeType="application/json" )) return resources except Exception as e: logger.error(f"Error listing quotes resources: {e}") return [] async def read_resource(self, uri: str, config: Optional[Dict[str, Any]] = None) -> Optional[ResourceContent]: """Read quote content with client-specific access control""" try: if not self.validate_uri(uri): raise ValueError(f"Invalid URI for quotes resource: {uri}") # Parse URI resource_path = uri.replace("quotes://", "") # Get configuration allowed_authors = [] search_enabled = True if config: allowed_authors = config.get("allowed_authors", []) search_enabled = config.get("search_enabled", True) if not self._db: raise ValueError("No database connection available") from .models import Quote async with self._db.get_session() as session: from sqlalchemy import select if resource_path.isdigit(): # Get specific quote by ID quote_id = int(resource_path) quote = await session.get(Quote, quote_id) if not quote: raise ValueError(f"Quote not found: {quote_id}") # Check author access if allowed_authors and quote.author not in allowed_authors: raise ValueError(f"Access denied to author: {quote.author}") content = f'"{quote.quote}" - {quote.author}' return ResourceContent( uri=uri, mimeType="text/plain", text=content ) elif resource_path == "search": # Return search instructions if not search_enabled: raise ValueError("Search is disabled for this client") search_info = { "description": "Search quotes by adding ?q=term to the URI", "examples": [ "quotes://search?q=inspiration", "quotes://search?q=Einstein", "quotes://search?q=success" ], "note": "Search looks in both quote content and author names" } return ResourceContent( uri=uri, mimeType="application/json", text=str(search_info) ) elif resource_path == "authors": # Get list of authors query = select(Quote.author).distinct() if allowed_authors: query = query.where(Quote.author.in_(allowed_authors)) result = await session.execute(query) authors = [row[0] for row in result.fetchall()] authors.sort() authors_info = { "total_authors": len(authors), "authors": authors } return ResourceContent( uri=uri, mimeType="application/json", text=str(authors_info) ) elif resource_path.startswith("search?q="): # Handle search query if not search_enabled: raise ValueError("Search is disabled for this client") # Extract search term search_term = resource_path.split("search?q=", 1)[1] search_term = search_term.replace("%20", " ") # Basic URL decoding # Search in quotes and authors query = select(Quote).where( (Quote.quote.ilike(f"%{search_term}%")) | (Quote.author.ilike(f"%{search_term}%")) ) if allowed_authors: query = query.where(Quote.author.in_(allowed_authors)) query = query.limit(20) # Limit search results result = await session.execute(query) quotes = result.scalars().all() search_results = { "search_term": search_term, "total_results": len(quotes), "quotes": [ { "id": quote.id, "author": quote.author, "quote": quote.quote, "uri": f"quotes://{quote.id}" } for quote in quotes ] } return ResourceContent( uri=uri, mimeType="application/json", text=str(search_results) ) else: raise ValueError(f"Unknown resource path: {resource_path}") except Exception as e: logger.error(f"Error reading quotes resource {uri}: {e}") raise