get_neighborhood

Instructions

Implementation Reference

• src/mcp_server/tools.py:417-513 (handler)

  The actual implementation of get_neighborhood. BFS over the resolved-link graph treating links as undirected. Uses NoteLink table to find neighbors up to depth hops (capped at 5), with a limit on distinct notes (capped at 200). Returns formatted markdown output showing distance, title, file path, tags, and via path for each neighbor.

  @_tracked("get_neighborhood", ["path", "depth", "limit"])
  async def get_neighborhood_impl(path: str, depth: int = 1, limit: int = 50) -> str:
      """BFS over the resolved-link graph treating links as undirected."""
      from sqlalchemy import or_, select
      from src.models.db import NoteLink, NoteMetadata
  
      uid = current_user_id.get()
      depth = max(1, min(depth, 5))
      limit = max(1, min(limit, 200))
  
      async with async_session() as session:
          src_stmt = select(NoteMetadata).where(NoteMetadata.file_path == path)
          if uid is not None:
              src_stmt = src_stmt.where(NoteMetadata.user_id == uid)
          source = (await session.execute(src_stmt)).scalar_one_or_none()
          if source is None:
              return f"Note not found: {path}"
  
          # BFS state.
          seen: dict[int, dict] = {source.id: {"distance": 0, "via": None}}
          frontier: list[int] = [source.id]
          truncated = False
  
          for d in range(1, depth + 1):
              if not frontier:
                  break
              stmt = select(
                  NoteLink.source_note_id,
                  NoteLink.target_note_id,
              ).where(
                  or_(
                      NoteLink.source_note_id.in_(frontier),
                      NoteLink.target_note_id.in_(frontier),
                  ),
                  NoteLink.target_note_id.isnot(None),
              )
              edges = (await session.execute(stmt)).all()
              next_frontier: list[int] = []
              for src_id, tgt_id in edges:
                  # Walk both directions.
                  for from_id, to_id in ((src_id, tgt_id), (tgt_id, src_id)):
                      if from_id in seen and to_id not in seen:
                          seen[to_id] = {"distance": d, "via": from_id}
                          next_frontier.append(to_id)
                          if len(seen) - 1 >= limit:
                              truncated = True
                              break
                  if truncated:
                      break
              frontier = next_frontier
              if truncated:
                  break
  
          # Hydrate metadata for everything except the source. The BFS edges
          # were already scoped to this user's graph (indexer guarantees the
          # vault_index is per-user), but we filter again here as a defense
          # in depth so a corrupted state can't leak rows across users.
          ids = [nid for nid in seen if nid != source.id]
          if not ids:
              return f"`{path}` has no resolved-link neighbors"
          meta_stmt = select(NoteMetadata).where(NoteMetadata.id.in_(ids))
          if uid is not None:
              meta_stmt = meta_stmt.where(NoteMetadata.user_id == uid)
          meta_rows = (await session.execute(meta_stmt)).scalars().all()
          meta_by_id = {m.id: m for m in meta_rows}
          # Drop any ids that the user_id filter excluded (shouldn't happen
          # under normal operation but keeps the output consistent).
          ids = [i for i in ids if i in meta_by_id]
          if not ids:
              return f"`{path}` has no resolved-link neighbors"
          # We also need `via` paths — fetch those.
          via_ids = {seen[nid]["via"] for nid in ids if seen[nid]["via"] is not None}
          via_paths = {source.id: source.file_path}
          if via_ids - {source.id}:
              via_stmt = select(NoteMetadata.id, NoteMetadata.file_path).where(
                  NoteMetadata.id.in_(via_ids)
              )
              if uid is not None:
                  via_stmt = via_stmt.where(NoteMetadata.user_id == uid)
              via_rows = (await session.execute(via_stmt)).all()
              for vid, vpath in via_rows:
                  via_paths[vid] = vpath
  
      ordered = sorted(ids, key=lambda nid: (seen[nid]["distance"], meta_by_id[nid].file_path))
      lines = [
          f"Neighborhood of `{path}` (depth ≤ {depth}, {len(ordered)} notes"
          + (", truncated" if truncated else "") + "):\n"
      ]
      for nid in ordered:
          m = meta_by_id[nid]
          info = seen[nid]
          via_path = via_paths.get(info["via"], "?")
          tags_str = f" [{', '.join(m.tags)}]" if m.tags else ""
          lines.append(
              f"- d={info['distance']} **{m.title}** (`{m.file_path}`){tags_str} via `{via_path}`"
          )
      return "\n".join(lines)

• src/mcp_server/server.py:271-286 (registration)

  MCP tool registration of get_neighborhood via @mcp.tool() decorator. Defines the tool signature (path, depth=1, limit=50) and delegates to get_neighborhood_impl.

  @mcp.tool()
  async def get_neighborhood(path: str, depth: int = 1, limit: int = 50) -> str:
      """The connected subgraph reachable from `path` via links or backlinks,
      up to `depth` hops (treated as undirected).
  
      Use this when an agent needs the local cluster around a topic — e.g.
      "summarize everything connected to this project". Prefer this over
      `find_related` when explicit links are the signal you want; prefer
      `find_related` when the connection is conceptual rather than linked.
  
      Args:
          path: Vault-relative path to the seed note.
          depth: Maximum BFS depth (default 1, capped at 5).
          limit: Maximum distinct neighbor notes (default 50, hard cap 200).
      """
      return await get_neighborhood_impl(path, depth=depth, limit=limit)

• src/mcp_server/tools.py:73-90 (helper)

  The _tracked decorator that wraps get_neighborhood_impl (and all other tools). Records timing, parameters, and response size to the usage_logs table.

  def _tracked(tool_name: str, param_keys: list[str]):
      """Decorator that times the call and logs it to usage_logs."""
      def decorator(fn):
          @wraps(fn)
          async def wrapper(*args, **kwargs):
              start = time.monotonic()
              result = await fn(*args, **kwargs)
              duration_ms = int((time.monotonic() - start) * 1000)
              params = {}
              for i, key in enumerate(param_keys):
                  if i < len(args):
                      params[key] = args[i]
                  elif key in kwargs:
                      params[key] = kwargs[key]
              await _log_usage(tool_name, _truncate_params(params), duration_ms, len(str(result)))
              return result
          return wrapper
      return decorator

• src/mcp_server/server.py:271-286 (schema)

  Input schema/parameters defined via the @mcp.tool() function signature: path (str), depth (int, default 1), limit (int, default 50). The docstring describes behavior and constraints.

  @mcp.tool()
  async def get_neighborhood(path: str, depth: int = 1, limit: int = 50) -> str:
      """The connected subgraph reachable from `path` via links or backlinks,
      up to `depth` hops (treated as undirected).
  
      Use this when an agent needs the local cluster around a topic — e.g.
      "summarize everything connected to this project". Prefer this over
      `find_related` when explicit links are the signal you want; prefer
      `find_related` when the connection is conceptual rather than linked.
  
      Args:
          path: Vault-relative path to the seed note.
          depth: Maximum BFS depth (default 1, capped at 5).
          limit: Maximum distinct neighbor notes (default 50, hard cap 200).
      """
      return await get_neighborhood_impl(path, depth=depth, limit=limit)