Add lightweight code context indexing

mcp_server/client.py

@@ -197,6 +197,73 @@ async def search(
     return "\n".join(lines_out)
 
 
+async def context(
+    query: str,
+    limit: int = 8,
+    include_snippets: bool = True,
+) -> str:
+    """Call POST /context and format a compact agent context bundle."""
+    client = await get_client()
+    response = await client.post(
+        "/context",
+        json={
+            "query": query,
+            "limit": limit,
+            "include_snippets": include_snippets,
+        },
+    )
+
+    if response.status_code != 200:
+        return _handle_error(response)
+
+    data = response.json()
+    symbols = data.get("symbols", [])
+    snippets = data.get("snippets", [])
+    related_documents = data.get("related_documents", [])
+
+    lines_out = [f"Context for '{query}'"]
+
+    if symbols:
+        lines_out.append("")
+        lines_out.append("Symbols:")
+        for s in symbols:
+            location = s.get("source_path") or s.get("filename")
+            lines_out.append(
+                f"  {s.get('qualified_name')} [{s.get('kind')}] "
+                f"{location}:{s.get('start_line')}-{s.get('end_line')}"
+            )
+            if s.get("signature"):
+                lines_out.append(f"    {s['signature']}")
+
+    if snippets:
+        lines_out.append("")
+        lines_out.append("Snippets:")
+        for snip in snippets:
+            location = snip.get("source_path") or snip.get("filename")
+            lines_out.append(
+                f"--- {location}:{snip.get('start_line')}-{snip.get('end_line')} "
+                f"({snip.get('symbol')})"
+            )
+            lines_out.append(snip.get("text", ""))
+
+    if related_documents:
+        lines_out.append("")
+        lines_out.append("Related documents:")
+        for doc in related_documents:
+            lines_out.append(
+                f"  {doc.get('filename')} page {doc.get('page_number')} "
+                f"score {doc.get('relevance_score')}"
+            )
+            if doc.get("highlight"):
+                lines_out.append(f"    {doc['highlight']}")
+
+    if not symbols and not related_documents:
+        lines_out.append("")
+        lines_out.append("No context found.")
+
+    return "\n".join(lines_out)
+
+
 async def get_info() -> str:
     """Call GET /info and format as readable text."""
     client = await get_client()

mcp_server/models.py

@@ -58,6 +58,20 @@ class SearchInput(BaseModel):
     offset: int = Field(0, description="Pagination offset", ge=0)
 
 
+class ContextInput(BaseModel):
+    """Input for compact agent-oriented context lookup."""
+
+    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
+
+    query: str = Field(
+        ..., description="Task, symbol, or topic to build compact context for", min_length=1
+    )
+    limit: int = Field(8, description="Max symbols/results", ge=1, le=20)
+    include_snippets: bool = Field(
+        True, description="Include small source snippets around matching symbols"
+    )
+
+
 class GlobInput(BaseModel):
     """Input for finding files matching a glob pattern."""
 

mcp_server/server.py

@@ -1,4 +1,4 @@
-"""OpenDB MCP Server — 3 tools: read, search, glob."""
+"""OpenDB MCP Server."""
 
 from __future__ import annotations
 
@@ -10,7 +10,7 @@
 
 from mcp_server.client import close_client
 from mcp_server.models import (
-    AddWorkspaceInput, CurrentWorkspaceInput, GlobInput, InfoInput,
+    AddWorkspaceInput, ContextInput, CurrentWorkspaceInput, GlobInput, InfoInput,
     ListWorkspacesInput, MemoryForgetInput, MemoryRecallInput, MemoryStoreInput,
     ReadInput, RemoveWorkspaceInput, SearchInput, UseWorkspaceInput,
 )
@@ -181,6 +181,30 @@ async def opendb_search(params: SearchInput) -> str:
     )
 
 
+@mcp.tool(
+    name="opendb_context",
+    annotations={
+        "title": "Build Context",
+        "readOnlyHint": True,
+        "destructiveHint": False,
+        "idempotentHint": True,
+        "openWorldHint": False,
+    },
+)
+async def opendb_context(params: ContextInput) -> str:
+    """Build compact context from indexed code symbols plus full-text hits.
+
+    Use this after opendb_search finds a concept, or directly when you know a
+    symbol/function/class name. It returns locations and small snippets instead
+    of dumping entire files into the agent context.
+    """
+    return await opendb.context(
+        query=params.query,
+        limit=params.limit,
+        include_snippets=params.include_snippets,
+    )
+
+
 @mcp.tool(
     name="opendb_glob",
     annotations={

opendb/cli.py

@@ -5,6 +5,7 @@
     opendb init [PATH]          # create .opendb/ in PATH (default: current dir)
     opendb index [PATH]         # index PATH (default: current dir)
     opendb search QUERY         # search indexed files
+    opendb context QUERY        # build compact code/document context
     opendb read FILENAME        # read a file
     opendb memory profile       # render a white-box memory profile
     opendb serve-mcp            # start MCP server (stdio, embedded mode)
@@ -136,6 +137,52 @@ async def _search() -> dict:
         typer.echo("")
 
 
+# ---------------------------------------------------------------------------
+# context
+# ---------------------------------------------------------------------------
+
+@app.command()
+def context(
+    query: str = typer.Argument(..., help="Task, symbol, or topic"),
+    workspace: Path = typer.Option(Path("."), "--workspace", "-w", help="Workspace root"),
+    limit: int = typer.Option(8, help="Maximum symbols/results"),
+    json_output: bool = typer.Option(False, "--json", help="Output as JSON"),
+) -> None:
+    """Build compact context from indexed code symbols and documents."""
+    from opendb_core.workspace import Workspace
+
+    ws = Workspace.open(workspace)
+
+    async def _context() -> dict:
+        await ws.init()
+        result = await ws.context(query, limit=limit)
+        await ws.close()
+        return result
+
+    result = _run(_context())
+
+    if json_output:
+        typer.echo(json.dumps(result, indent=2, ensure_ascii=False))
+        return
+
+    typer.echo(f"Context for '{query}':\n")
+    for s in result.get("symbols", []):
+        location = s.get("source_path") or s.get("filename")
+        typer.echo(
+            f"  {s.get('qualified_name')} [{s.get('kind')}] "
+            f"{location}:{s.get('start_line')}-{s.get('end_line')}"
+        )
+    if result.get("snippets"):
+        typer.echo("")
+        for snip in result["snippets"]:
+            location = snip.get("source_path") or snip.get("filename")
+            typer.echo(
+                f"--- {location}:{snip.get('start_line')}-{snip.get('end_line')}"
+            )
+            typer.echo(snip.get("text", ""))
+            typer.echo("")
+
+
 # ---------------------------------------------------------------------------
 # read
 # ---------------------------------------------------------------------------

opendb_core/main.py

@@ -9,6 +9,7 @@
 from opendb_core.config import settings
 from opendb_core.services.watch_service import stop_all as stop_all_watchers
 from opendb_core.routers.files import router as files_router
+from opendb_core.routers.context import router as context_router
 from opendb_core.routers.glob import router as glob_router
 from opendb_core.routers.health import router as health_router
 from opendb_core.routers.info import router as info_router
@@ -89,6 +90,7 @@ async def value_error_handler(request: Request, exc: ValueError) -> JSONResponse
 app.include_router(health_router)
 app.include_router(info_router)
 app.include_router(files_router)
+app.include_router(context_router)
 app.include_router(glob_router)
 app.include_router(index_router)
 app.include_router(read_router)

opendb_core/routers/context.py

@@ -0,0 +1,26 @@
+"""Context endpoint: compact agent-oriented search bundle."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+from opendb_core.services.context_service import build_context
+
+router = APIRouter(tags=["context"])
+
+
+class ContextRequest(BaseModel):
+    query: str = Field(..., min_length=1)
+    limit: int = Field(8, ge=1, le=20)
+    include_snippets: bool = True
+
+
+@router.post("/context")
+async def context(request: ContextRequest) -> dict:
+    """Build compact code/document context for an agent task or symbol lookup."""
+    return await build_context(
+        query=request.query,
+        limit=request.limit,
+        include_snippets=request.include_snippets,
+    )

opendb_core/services/context_service.py

@@ -0,0 +1,59 @@
+"""Compact context builder for agent code/document exploration."""
+
+from __future__ import annotations
+
+from opendb_core.storage import get_backend
+from opendb_core.utils.text import extract_lines
+
+
+async def build_context(query: str, limit: int = 8, include_snippets: bool = True) -> dict:
+    """Build a small context bundle from symbols plus full-text hits."""
+    backend = get_backend()
+    symbol_limit = max(1, min(limit, 20))
+    symbols = await backend.search_code_symbols(query, limit=symbol_limit)
+
+    snippets: list[dict] = []
+    if include_snippets:
+        seen: set[tuple[str, int, int]] = set()
+        for symbol in symbols[: min(symbol_limit, 8)]:
+            file_id = str(symbol["file_id"])
+            start = max(1, int(symbol["start_line"]) - 1)
+            end = int(symbol["end_line"]) + 1
+            key = (file_id, start, end)
+            if key in seen:
+                continue
+            seen.add(key)
+            text_row = await backend.get_file_text(file_id)
+            snippets.append({
+                "file_id": file_id,
+                "filename": symbol["filename"],
+                "source_path": symbol.get("source_path"),
+                "symbol": symbol["qualified_name"],
+                "start_line": start,
+                "end_line": end,
+                "text": extract_lines(
+                    text_row["full_text"],
+                    text_row["line_index"],
+                    start,
+                    min(end, int(text_row["total_lines"])),
+                ),
+            })
+
+    doc_hits = await backend.search_fts(query, {}, max(1, min(limit, 10)), 0)
+    symbol_file_ids = {str(s["file_id"]) for s in symbols}
+    related_documents = [
+        hit for hit in doc_hits.get("results", [])
+        if str(hit.get("file_id")) not in symbol_file_ids
+    ][: max(0, limit - len(symbols[:limit]))]
+
+    return {
+        "query": query,
+        "symbols": symbols[:limit],
+        "snippets": snippets,
+        "related_documents": related_documents,
+        "stats": {
+            "symbol_count": len(symbols),
+            "snippet_count": len(snippets),
+            "related_document_count": len(related_documents),
+        },
+    }

opendb_core/storage/base.py

@@ -137,6 +137,16 @@ async def search_fts(
         """
         ...
 
+    async def search_code_symbols(
+        self,
+        query: str,
+        *,
+        limit: int = 20,
+        kinds: list[str] | None = None,
+    ) -> list[dict]:
+        """Search indexed code symbols by name/signature/docstring."""
+        ...
+
     # ------------------------------------------------------------------
     # Indexing
     # ------------------------------------------------------------------