"""
The tools that answer from the index of the repositories and the blog.
- ``get_repository_overview``: what a repository contains, its layout, and its README.
- ``search_documentation``: ranked passages of the documentation.
- ``search_code``: ranked files of source code, with their matching lines.
- ``read_file``: a file, or a range of its lines, with line numbers.
- ``list_files``: the folders and files under a folder.
- ``search_blog``: ranked passages of the posts of the blog, blog.smarter.sh.
- ``read_blog_post``: a post of the blog, with line numbers.
The blog's tools have no ``repository`` argument, and only the MCP server of all of the
repositories serves them. A tool that is called on a repository that is not indexed yet
says so, and queues the repository for indexing.
.. note::
**Experimental.** The MCPServer was designed and coded by Claude Code (Anthropic's
Claude Opus 5.5), with Lawrence McDaniel as co-author. It is experimental, and will
be documented.
"""
import posixpath
from typing import ClassVar, Optional, Sequence, cast
from cache_results import lazy_cache
from pydantic import BaseModel, ConfigDict, Field
from smarter.apps.mcpserver.exceptions import SmarterMCPServerToolError
from smarter.apps.mcpserver.index.search import SearchHit, search_files
from smarter.apps.mcpserver.models import (
RepositoryFile,
RepositoryFileKind,
RepositoryIndex,
)
from smarter.apps.mcpserver.protocol.tools import (
DEFAULT_SEARCH_LIMIT,
MAX_READ_CHARACTERS,
MAX_SEARCH_LIMIT,
SmarterProjectTool,
ToolArguments,
)
from smarter.apps.mcpserver.repositories import (
SMARTER_PROJECT_DOCS_LATEST_URL,
ProjectBlog,
ProjectRepository,
ProjectSource,
project_blog,
)
MAX_READ_LINES = 400
"""The maximum number of lines that read_file returns at once."""
MAX_LIST_ENTRIES = 300
MAX_README_CHARACTERS = 6000
README_NAMES = ("README.md", "README.rst", "readme.md", "README.txt", "README")
INDEX_QUEUE_TIMEOUT = 600
"""Seconds during which a repository that is not indexed is queued for indexing only once."""
CODE_FENCE_LANGUAGES = {
".py": "python",
".ts": "typescript",
".tsx": "tsx",
".js": "javascript",
".jsx": "jsx",
".go": "go",
".json": "json",
".yaml": "yaml",
".yml": "yaml",
".toml": "toml",
".sh": "bash",
".html": "html",
".css": "css",
".scss": "scss",
".tf": "hcl",
".hcl": "hcl",
".sql": "sql",
".rst": "rst",
".md": "markdown",
}
[docs]
def queue_indexing_key(name: str) -> str:
"""Return the cache key that remembers that a repository was queued for indexing."""
return f"smarter.apps.mcpserver.index.tools.queue_indexing.{name}"
def queue_indexing(repository: ProjectSource) -> None:
"""
Queue a repository, or the blog, that is not indexed for indexing, at most once every :data:`INDEX_QUEUE_TIMEOUT` seconds.
:param repository: The repository, or the blog.
"""
# pylint: disable=import-outside-toplevel
from smarter.apps.mcpserver.tasks import index_blog, index_repository
if lazy_cache.add(queue_indexing_key(repository.name), True, timeout=INDEX_QUEUE_TIMEOUT):
if isinstance(repository, ProjectBlog):
index_blog.delay()
else:
index_repository.delay(repository.name)
def ready_indexes(repositories: Sequence[ProjectSource]) -> dict[str, RepositoryIndex]:
"""
Return the indexes of the repositories that are ready to search, and queue the others for indexing.
:param repositories: The repositories, or the blog.
:returns: The ready indexes, by repository name.
"""
indexes = {
index.repository: index
for index in RepositoryIndex.objects.filter(repository__in=[r.name for r in repositories])
}
ready = {}
for repository in repositories:
index = indexes.get(repository.name)
if index is not None and index.is_ready:
ready[repository.name] = index
else:
queue_indexing(repository)
return ready
def require_index(repository: ProjectRepository) -> RepositoryIndex:
"""
Return a repository's index.
:param repository: The repository.
:returns: Its index.
:raises SmarterMCPServerToolError: If the repository is not indexed yet. It is then queued for indexing.
"""
index = ready_indexes([repository]).get(repository.name)
if index is None:
raise SmarterMCPServerToolError(
f"{repository.name} has not been indexed yet. Indexing has started: try again in a few minutes."
)
return index
def not_indexed_note(repositories: list[ProjectRepository], ready: dict[str, RepositoryIndex]) -> str:
"""Return a note that names the repositories that were not searched because they are not indexed yet."""
missing = [repository.name for repository in repositories if repository.name not in ready]
if not missing:
return ""
return f"\n\nNot searched, because they are still being indexed: {', '.join(missing)}. Try again in a few minutes."
def code_fence_language(path: str) -> str:
"""Return the Markdown code fence language of a file."""
return CODE_FENCE_LANGUAGES.get(posixpath.splitext(path)[1].lower(), "text")
# -----------------------------------------------------------------------------
# get_repository_overview
# -----------------------------------------------------------------------------
# -----------------------------------------------------------------------------
# search_documentation, search_code
# -----------------------------------------------------------------------------
class SearchArguments(ToolArguments):
"""The arguments of search_documentation."""
query: str = Field(
min_length=1,
max_length=500,
description='Keywords, identifiers or "quoted phrases", e.g. custom domain verification.',
)
limit: int = Field(
default=DEFAULT_SEARCH_LIMIT, ge=1, le=MAX_SEARCH_LIMIT, description="The maximum number of results."
)
class SearchCodeArguments(SearchArguments):
"""The arguments of search_code."""
path: Optional[str] = Field(
default=None,
max_length=512,
description="Only search files under this folder, e.g. smarter/smarter/apps/llmclient.",
)
class SearchTool(SmarterProjectTool):
"""The base class of the search tools."""
kind: ClassVar[str]
noun: ClassVar[str]
def search(self, arguments: SearchArguments, path: Optional[str] = None) -> str:
"""Search the scope's repositories, and return the hits as Markdown."""
repository = self.scope.resolve(arguments.repository, required=False)
repositories = [repository] if repository else list(self.scope.repositories)
ready = ready_indexes(repositories)
if not ready:
raise SmarterMCPServerToolError(
f"{', '.join(r.name for r in repositories)} has not been indexed yet. "
"Indexing has started: try again in a few minutes."
)
hits = search_files(
[r for r in repositories if r.name in ready],
arguments.query,
kind=self.kind,
path=path,
limit=arguments.limit,
)
where = repository.name if repository else "The Smarter Project's repositories"
if not hits:
return (
f'No {self.noun} of {where} matches "{arguments.query}". Try other keywords, an identifier, '
"or the other search tool." + not_indexed_note(repositories, ready)
)
results = "\n\n".join(self.format_hit(number, hit) for number, hit in enumerate(hits, start=1))
return (
f'{len(hits)} results in the {self.noun} of {where}, for "{arguments.query}", best first. '
f"Read a file with read_file to see more of it.\n\n{results}" + not_indexed_note(repositories, ready)
)
def format_hit(self, number: int, hit: SearchHit) -> str:
"""Return one hit, as Markdown."""
heading = f"{number}. **{hit.title}** `{hit.path}`" if hit.title else f"{number}. `{hit.path}`"
if not self.scope.is_single:
heading += f" in {hit.repository}"
links = []
if hit.documentation_url:
links.append(f" Documentation: {hit.documentation_url}")
links.append(f" Source: {hit.url}")
width = len(str(hit.lines[-1].number)) if hit.lines else 1
excerpt = "\n".join(f"{line.number:>{width}} {line.text}" for line in hit.lines)
fence = code_fence_language(hit.path) if hit.kind == RepositoryFileKind.CODE else "text"
return f"{heading}\n" + "\n".join(links) + f"\n\n```{fence}\n{excerpt}\n```"
# -----------------------------------------------------------------------------
# read_file
# -----------------------------------------------------------------------------
class ReadLines(BaseModel):
"""The lines of a file that :func:`read_lines` selected."""
model_config = ConfigDict(frozen=True, extra="forbid")
start: int = Field(description="The first line, from 1.")
end: int = Field(description="The last line.")
total: int = Field(description="The file's number of lines.")
text: str = Field(description="The lines, each prefixed with its number.")
def read_lines(content: str, start_line: int, end_line: Optional[int]) -> ReadLines:
"""
Return a range of a file's lines, with line numbers, within :data:`MAX_READ_LINES` and :data:`MAX_READ_CHARACTERS`.
:param content: The file's content.
:param start_line: The first line to read, from 1.
:param end_line: The last line to read, or None to read as many as allowed.
:returns: The lines.
"""
lines = content.splitlines()
total = len(lines)
start = min(start_line, max(total, 1))
end = min(end_line or start + MAX_READ_LINES - 1, start + MAX_READ_LINES - 1, total)
width = len(str(end))
selected: list[str] = []
characters = 0
for number in range(start, end + 1):
text = f"{number:>{width}} {lines[number - 1]}"
characters += len(text) + 1
if characters > MAX_READ_CHARACTERS and selected:
end = number - 1
break
selected.append(text)
return ReadLines(start=start, end=end, total=total, text="\n".join(selected))
class ReadFileArguments(ToolArguments):
"""The arguments of read_file."""
path: str = Field(min_length=1, max_length=512, description="The file's path in the repository, e.g. README.md.")
start_line: int = Field(default=1, ge=1, description="The first line to read.")
end_line: Optional[int] = Field(
default=None, ge=1, description=f"The last line to read. At most {MAX_READ_LINES} lines are read at once."
)
# -----------------------------------------------------------------------------
# list_files
# -----------------------------------------------------------------------------
class ListFilesArguments(ToolArguments):
"""The arguments of list_files."""
path: Optional[str] = Field(
default=None,
max_length=512,
description="The folder, e.g. smarter/smarter/apps. Omit it for the repository's top-level folder.",
)
# -----------------------------------------------------------------------------
# search_blog, read_blog_post
# -----------------------------------------------------------------------------
def require_blog_index(blog: ProjectBlog) -> RepositoryIndex:
"""
Return the blog's index.
:param blog: The blog.
:returns: Its index.
:raises SmarterMCPServerToolError: If the blog is not indexed yet. It is then queued for indexing.
"""
index = ready_indexes([blog]).get(blog.name)
if index is None:
raise SmarterMCPServerToolError(
f"The blog, {blog.url}, has not been indexed yet. Indexing has started: try again in a few minutes."
)
return index
[docs]
class SearchBlogTool(SmarterProjectTool):
"""Search the blog's posts."""
name = "search_blog"
title = "Search blog"
description = (
"Search the posts of The Smarter Project's blog, blog.smarter.sh: release announcements, with what "
"changed in each version and how to upgrade, and articles about how and why Smarter was built, e.g. "
"its security, logging, testing, caching and use of AI coding agents. Returns the best passages, with "
"links to the posts. Use it, together with search_documentation, for why and how questions, design "
"decisions, what changed in a release, and the project's history."
)
arguments_model = SearchArguments
uses_repository = False
project_wide = True
[docs]
def run(self, arguments: ToolArguments) -> str:
arguments = cast(SearchArguments, arguments)
blog = project_blog()
require_blog_index(blog)
hits = search_files([blog], arguments.query, kind=RepositoryFileKind.DOCUMENTATION, limit=arguments.limit)
if not hits:
return (
f'No post of the blog, {blog.url}, matches "{arguments.query}". Try other keywords, or '
"search_documentation."
)
results = "\n\n".join(self.format_hit(number, hit) for number, hit in enumerate(hits, start=1))
return (
f'{len(hits)} posts of the blog, {blog.url}, for "{arguments.query}", best first. '
f"Read a post with read_blog_post to see more of it.\n\n{results}"
)
[docs]
@staticmethod
def format_hit(number: int, hit: SearchHit) -> str:
"""Return one hit, as Markdown."""
width = len(str(hit.lines[-1].number)) if hit.lines else 1
excerpt = "\n".join(f"{line.number:>{width}} {line.text}" for line in hit.lines)
return f"{number}. **{hit.title or hit.path}** `{hit.path}`\n Post: {hit.url}\n\n```text\n{excerpt}\n```"
class ReadBlogPostArguments(ToolArguments):
"""The arguments of read_blog_post."""
path: str = Field(
min_length=1,
max_length=512,
description="The post's path, e.g. 2026/10/26/release-smarter-api-v0-6, or its url.",
)
start_line: int = Field(default=1, ge=1, description="The first line to read.")
end_line: Optional[int] = Field(
default=None, ge=1, description=f"The last line to read. At most {MAX_READ_LINES} lines are read at once."
)
[docs]
class ReadBlogPostTool(SmarterProjectTool):
"""Read a post of the blog."""
name = "read_blog_post"
title = "Read blog post"
description = (
"Read a post of The Smarter Project's blog, blog.smarter.sh, as text with line numbers, from the "
f"paths or urls that search_blog returns. Long posts are read {MAX_READ_LINES} lines at a time: use "
"start_line and end_line to read more."
)
arguments_model = ReadBlogPostArguments
uses_repository = False
project_wide = True
[docs]
def run(self, arguments: ToolArguments) -> str:
arguments = cast(ReadBlogPostArguments, arguments)
blog = project_blog()
index = require_blog_index(blog)
path = blog.post_path(arguments.path)
row = RepositoryFile.objects.filter(index=index, path=path).values("content").first()
if row is None:
slug = path.rsplit("/", 1)[-1]
suggestions = list(
RepositoryFile.objects.filter(index=index, path__icontains=slug)
.order_by("-path")
.values_list("path", flat=True)[:8]
)
message = f"{path} is not a post of the blog, {blog.url}."
if suggestions and slug:
message += " Did you mean: " + ", ".join(suggestions) + "?"
else:
message += " Use search_blog to find it."
raise SmarterMCPServerToolError(message)
lines = read_lines(row["content"], arguments.start_line, arguments.end_line)
header = f"{blog.post_url(path)}, lines {lines.start}-{lines.end} of {lines.total}"
footer = (
f"\n\nThe post continues: call read_blog_post with start_line={lines.end + 1}."
if lines.end < lines.total
else ""
)
return f"{header}\n\n```markdown\n{lines.text}\n```{footer}"
__all__ = [
"GetRepositoryOverviewTool",
"ListFilesTool",
"ReadBlogPostTool",
"ReadFileTool",
"SearchBlogTool",
"SearchCodeTool",
"SearchDocumentationTool",
"queue_indexing_key",
]