MCP search_schemas tool

.github/workflows/ci-cd.yml

@@ -41,7 +41,7 @@ jobs:
       - name: Install Ruff
         run: |
           python -m pip install --upgrade pip
-          pip install ruff
+          pip install "ruff>=0.15,<0.16"
 
       - name: Lint
         run: ruff check --output-format github
@@ -57,7 +57,7 @@ jobs:
       - name: Install Ruff
         run: |
           python -m pip install --upgrade pip
-          pip install ruff
+          pip install "ruff>=0.15,<0.16"
 
       - name: Lint
         # --output-format github requires --preview

core/api/views.py

@@ -39,9 +39,7 @@ def docs(request):
 @require_GET
 def find(request):
     id_value = request.GET.get("id")
-    published_schema_refs = SchemaRef.objects.filter(
-        schema__in=Schema.public_objects.all()
-    )
+    published_schema_refs = SchemaRef.objects.filter(schema__in=Schema.objects.public())
     schema_ref = get_object_or_404(published_schema_refs, id_value__iexact=id_value)
     return ApiResponse({"url": schema_ref.url})
 

core/forms.py

@@ -83,13 +83,13 @@ def clean_url(self):
         # If the schema is unpublished, we don't care if the URL or $id are already in use
         if (
             self.schema_id is None
-            or not Schema.public_objects.filter(id=self.schema_id).exists()
+            or not Schema.objects.public().filter(id=self.schema_id).exists()
         ):
             return url
 
         # But if it's a published schema, we need to make sure the URL and $id aren't already in use
         schema_refs = SchemaRef.objects.select_related("schema").filter(
-            schema__in=Schema.public_objects.exclude(id=self.schema_id)
+            schema__in=Schema.objects.public().exclude(id=self.schema_id)
         )
         # First check the URL
         for schema_ref in schema_refs:

core/mcp/server.py

@@ -1,8 +1,8 @@
 import json
+from typing import Literal
 from jsonschema import ValidationError as JSONValidationError
 from django.core.exceptions import ValidationError as DjangoValidationError
 from django.urls import reverse
-from django.db.models import Q
 from django.utils import timezone
 from mcp.server.fastmcp import FastMCP
 from core.models import Schema
@@ -16,12 +16,15 @@
 
 def format_schema(schema):
     formatted_schema = f"""
-Name: {schema.name}
 ID: {schema.id}
+Name: {schema.name}
 """
     if schema.description:
-        formatted_schema += f"""Description: {schema.description}
-"""
+        formatted_schema += f"Description: {schema.description}\n"
+
+    if schema.published_at is None or schema.published_at > timezone.now():
+        formatted_schema += "Visibility: Private\n"
+
     return formatted_schema
 
 
@@ -36,15 +39,70 @@ def ensure_current_user():
 
 # Note: We don't use type hints elsewhere in the codebase,
 # but they can influence FastMCP's behavior for tools and resources.
+# Function descriptions are the actual descriptions surfaced to models.
+
+MAX_PAGE_SIZE = 10
 
 
-# TODO: This will need to be paginated
 @mcp.tool()
 @sync_to_async
-def list_schemas():
-    """List all available schemas."""
-    public_schemas = [format_schema(schema) for schema in Schema.public_objects.all()]
-    return "\n---\n".join(public_schemas)
+def search_schemas(
+    query: str = None, scope: Literal["all", "user"] = "all", page: int = 1
+):
+    """
+    Search for schemas.
+
+    Args:
+      query: A search query. Can be a list of keywords or an $id. Pass None or an empty string to list all schemas in scope.
+      scope: 'user' to search only the user's own schemas (including private), or 'all' to search the entire registry. Defaults to 'all.'
+      page: Which page of search results to return. Defaults to 1.
+    """
+
+    user = ensure_current_user()
+
+    scope_results = (
+        Schema.objects.accessible_to(user)
+        if scope == "all"
+        else Schema.objects.filter(created_by=user)
+    )
+
+    matched_by_id_value = scope_results.filter(schemaref__id_value__iexact=query)
+
+    # If there is a query and it matches an exact ID, skip the full-text search.
+    if query and matched_by_id_value.exists():
+        results = matched_by_id_value
+    else:
+        results = scope_results.search(query)
+
+    total_count = results.count()
+    if total_count == 0:
+        return "No results matched your query."
+
+    total_pages = (total_count + MAX_PAGE_SIZE - 1) // MAX_PAGE_SIZE
+    if page < 1 or page > total_pages:
+        raise ValueError(
+            f"Invalid page number for query. Please request a page between 1 and {total_pages}."
+        )
+
+    start = (page - 1) * MAX_PAGE_SIZE
+    end = start + MAX_PAGE_SIZE
+    paginated_results = results[start:end]
+
+    formatted_results = [format_schema(schema) for schema in paginated_results]
+    formatted_page = "\n---\n".join(formatted_results)
+
+    response = f"Found {total_count} schema{'s' if total_count > 1 else ''} matching your query{':' if total_pages == 1 else '.'}"
+
+    if total_pages == 1:
+        response += f"\n\n{formatted_page}"
+        return response
+
+    response += f"\n\nThe results are truncated. Showing page {page} of {total_pages}:"
+    response += f"\n\n{formatted_page}"
+
+    response += f'\n\nTo get the next page, use `search_schemas(query: <keywords>, scope: "{scope}", page: {page + 1})'
+
+    return response
 
 
 @mcp.resource("schema://manifest.json")
@@ -61,20 +119,15 @@ async def get_schema(schema_id: int):
 
     @sync_to_async
     def fetch_from_db():
-        # TODO: dedupe from core.views:lookup_schema
-        schema_filter = Q(published_at__lte=timezone.now())
-
-        if user and user.is_authenticated:
-            schema_filter |= Q(created_by=user)
-
         try:
             schema = (
                 Schema.objects
+                .accessible_to(user)
                 .prefetch_related("schemaref_set")
                 .prefetch_related("documentationitem_set")
-                .filter(schema_filter)
                 .get(pk=schema_id)
             )
+
             return schema.to_manifest()
         except Schema.DoesNotExist:
             return None

core/models.py

@@ -1,6 +1,7 @@
 import logging
 from itertools import chain
 from django.db import models, transaction
+from django.db.models import Q
 from django.contrib.auth.models import User
 from django.utils import timezone
 from django.conf import settings
@@ -96,6 +97,10 @@ class Meta:
 
 
 class SchemaQuerySet(models.QuerySet):
+    def _get_public_q(self):
+        """Helper method to return the Q object for public schemas."""
+        return Q(published_at__isnull=False, published_at__lte=timezone.now())
+
     def search(self, query_text):
         """
         Rank schemas by full-text relevance against `query_text`.
@@ -117,14 +122,18 @@ def search(self, query_text):
             .order_by("-rank", "name")
         )
 
+    def public(self):
+        """Returns only public schemas."""
+        return self.filter(self._get_public_q())
 
-class PublicSchemaManager(models.Manager.from_queryset(SchemaQuerySet)):
-    def get_queryset(self):
-        return (
-            super()
-            .get_queryset()
-            .filter(published_at__isnull=False, published_at__lte=timezone.now())
-        )
+    def accessible_to(self, user):
+        """Returns schemas that are either public OR created by the user."""
+        q_filter = self._get_public_q()
+
+        if user and user.is_authenticated:
+            q_filter |= Q(created_by=user)
+
+        return self.filter(q_filter)
 
 
 class PublishedSchemaConflictError(Exception):
@@ -142,8 +151,7 @@ def __init__(self, conflicting_schema_ref, reason):
 
 
 class Schema(BaseModel):
-    objects = models.Manager()
-    public_objects = PublicSchemaManager()
+    objects = SchemaQuerySet.as_manager()
     name = models.CharField(max_length=200)
     published_at = models.DateTimeField(blank=True, null=True)
     permanent_urls = GenericRelation(PermanentURL, related_query_name="schema")
@@ -265,7 +273,7 @@ def check_for_published_conflicts(self):
             PublishedSchemaConflictError: If a conflict is found.
         """
         published_schema_refs = SchemaRef.objects.filter(
-            schema__in=Schema.public_objects.all()
+            schema__in=Schema.objects.public()
         ).exclude(
             # We don't want to check against this Schema's own SchemaRefs
             schema=self
@@ -826,7 +834,7 @@ def __str__(self):
 
     @property
     def public_schemas(self):
-        return Schema.public_objects.filter(
+        return Schema.objects.public().filter(
             created_by_id__in=self.profile_set.values_list("user_id", flat=True)
         )
 

core/views.py

@@ -2,7 +2,6 @@
 import uuid
 
 from django.shortcuts import render, get_object_or_404, redirect
-from django.db.models import Q
 from django.utils.html import escape
 from django.utils.safestring import mark_safe
 from django.contrib.auth.decorators import login_required
@@ -78,16 +77,11 @@
 def lookup_schema(function):
     @wraps(function)
     def _wrap_request(request, schema_id, *args, **kwargs):
-        schema_filter = Q(published_at__lte=timezone.now())
-
-        if request.user.is_authenticated:
-            schema_filter |= Q(created_by=request.user)
-
         schema = get_object_or_404(
             Schema.objects
+            .accessible_to(request.user)
             .prefetch_related("schemaref_set")
-            .prefetch_related("documentationitem_set")
-            .filter(schema_filter),
+            .prefetch_related("documentationitem_set"),
             pk=schema_id,
         )
 
@@ -127,7 +121,8 @@ def render_markdown(markdown_source_text):
 
 def index(request):
     defined_schemas = (
-        Schema.public_objects
+        Schema.objects
+        .public()
         .prefetch_related("schemaref_set")
         .exclude(schemaref__isnull=True)
         .order_by("name")
@@ -452,7 +447,7 @@ def organization_detail(request, organization_id):
 @login_required
 def manage_schema_permanent_urls(request, schema_id):
     schema = get_object_or_404(
-        Schema.public_objects,
+        Schema.objects.public(),
         id=schema_id,
         created_by=request.user,
     )