related_code_intel

Instructions

Implementation Reference

• src/project_code_intelligence/server.py:343-393 (handler)

  The handler function for the related_code_intel tool. It builds SQL queries to find code intelligence graph edges (edges from project_code_intel_edges) that match a record_id and/or symbol, filtered by collection, repo, and snapshot scope, ordered by id DESC, limited to 1-100 results.

  def tool_related_code_intel(args: Json) -> Json:
      record_id = optional_text(args, "record_id")
      symbol = optional_text(args, "symbol")
      if not record_id and not symbol:
          raise McpProtocolError("record_id or symbol is required")
      limit = require_int(args, "limit", 20, 1, 100)
      collection = scoped_collection(args)
      repo = optional_text(args, "repo")
  
      clauses = ["TRUE"]
      params: QueryParams = []
      if record_id:
          clauses.append("(e.source_record_id = %s OR e.target_record_id = %s)")
          params.extend([record_id, record_id])
      if symbol:
          clauses.append("(e.source_symbol = %s OR e.target_symbol = %s)")
          params.extend([symbol, symbol])
      if repo:
          clauses.append("e.repo = %s")
          params.append(repo)
      if collection:
          clauses.append("e.collection = %s")
          params.append(collection)
      snapshot_clauses, snapshot_params = scoped_snapshot_clauses(args, "e")
      clauses.extend(snapshot_clauses)
      params.extend(snapshot_params)
      params.append(limit)
  
      with db.connect() as conn:
          if not code_intel_tables_exist(conn):
              return ok({"error": "code intelligence schema is not initialized"})
          edges = conn.execute(
              db.query_sql(
                  query_with_where(
                      """
              SELECT e.id, e.snapshot_id, e.collection, e.repo, e.commit_sha,
                     e.source_record_id, e.target_record_id, e.edge_type,
                     e.source_symbol, e.target_symbol, e.source_path, e.target_path,
                     e.confidence_kind, e.metadata
              FROM project_code_intel_edges e
              """,
                      clauses,
                      """
              ORDER BY e.id DESC
              LIMIT %s
              """,
                  )
              ),
              params,
          ).fetchall()
      return ok({**snapshot_scope_response(args), "edges": edges})

• src/project_code_intelligence/mcp_tool_catalog.py:93-108 (schema)

  The ToolDefinition (metadata + JSON input schema) for related_code_intel. The schema accepts optional fields: record_id, symbol, collection, repo, limit (1-100), snapshot_id, include_historical.

  "related_code_intel": ToolDefinition(
      "Return code intelligence graph edges related to a record id or symbol.",
      {
          "type": "object",
          "properties": {
              "record_id": {"type": "string"},
              "symbol": {"type": "string"},
              "collection": {"type": "string"},
              "repo": {"type": "string"},
              "limit": {"type": "integer", "minimum": 1, "maximum": 100},
              "snapshot_id": {"type": "integer", "minimum": 1},
              "include_historical": {"type": "boolean"},
          },
          "additionalProperties": False,
      },
  ),

• src/project_code_intelligence/server.py:522-531 (registration)

  The TOOLS registry dict that maps the string 'related_code_intel' to its ToolDefinition and handler function (tool_related_code_intel).

  TOOLS: ToolRegistry = {
      "code_intel_status": (TOOL_DEFINITIONS["code_intel_status"], tool_code_intel_status),
      "search_code_intel_text": (TOOL_DEFINITIONS["search_code_intel_text"], tool_search_code_intel_text),
      "search_code_intel_semantic": (TOOL_DEFINITIONS["search_code_intel_semantic"], tool_search_code_intel_semantic),
      "get_code_intel_record": (TOOL_DEFINITIONS["get_code_intel_record"], tool_get_code_intel_record),
      "related_code_intel": (TOOL_DEFINITIONS["related_code_intel"], tool_related_code_intel),
      "search_static_findings": (TOOL_DEFINITIONS["search_static_findings"], tool_search_static_findings),
      "get_static_finding": (TOOL_DEFINITIONS["get_static_finding"], tool_get_static_finding),
      "get_static_code_flow": (TOOL_DEFINITIONS["get_static_code_flow"], tool_get_static_code_flow),
  }

• src/project_code_intelligence/mcp_protocol.py:69-77 (helper)

  The scoped_collection helper function used by the handler to determine the effective collection scope, checking configured vs requested collection with override permissions.

  def scoped_collection(args: Json) -> str | None:
      requested = optional_text(args, "collection")
      configured = config.configured_collection()
      if configured and requested and requested != configured and not config.collection_override_allowed():
          raise McpWritePermissionError(
              "collection override is disabled by PROJECT_CODE_INTELLIGENCE_COLLECTION; "
              "set PROJECT_CODE_INTELLIGENCE_ALLOW_COLLECTION_OVERRIDE=1 for trusted multi-collection access"
          )
      return requested or configured

• src/project_code_intelligence/mcp_filters.py:106-112 (helper)

  The scoped_snapshot_clauses helper used by the handler to build snapshot-related SQL WHERE clauses based on snapshot_id or include_historical flags.

  def scoped_snapshot_clauses(args: Json, alias: str) -> tuple[list[str], QueryParams]:
      scope, snapshot_id = snapshot_scope(args)
      if snapshot_id is not None:
          return [f"{column(alias, 'snapshot_id')} = %s"], [snapshot_id]
      if scope == "historical":
          return [], []
      return [latest_record_snapshot_clause(alias)], []