search_code_intel_text
Search or list code intelligence records using PostgreSQL full-text search and exact filters for collections, repos, symbols, metadata, and more.
Instructions
Search or list code intelligence records with optional PostgreSQL full-text search and exact filters.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | No | ||
| limit | No | ||
| collection | No | ||
| repo | No | ||
| record_type | No | ||
| language | No | ||
| file_role | No | ||
| content_class | No | ||
| confidence_kind | No | ||
| source_path | No | ||
| symbol | No | ||
| metadata_key | No | ||
| metadata_value | No | ||
| metadata_contains | No | ||
| snapshot_id | No | ||
| include_historical | No |
Implementation Reference
- Handler function for search_code_intel_text. Performs full-text search on project_code_intel_records using PostgreSQL tsquery when a query is provided, or lists records ordered by updated_at when no query is given. Supports optional filters via code_intel_clauses.
def tool_search_code_intel_text(args: Json) -> Json: query = optional_text(args, "query") limit = require_int(args, "limit", 10, 1, 50) clauses, filter_params = code_intel_clauses(args, "r") if query: clauses.append("r.search_document @@ websearch_to_tsquery('english', %s)") query_sql = query_with_where( """ SELECT r.id, r.snapshot_id, r.collection, r.repo, r.repo_role, r.branch, r.commit_sha, r.tree_sha, r.source_path, r.language, r.file_role, r.content_class, r.record_type, r.record_id, r.parent_record_id, r.title, r.summary, r.line_start, r.line_end, r.symbol, r.symbol_kind, r.confidence_kind, r.confidence, r.tool, r.rule_id, r.severity, r.metadata, r.updated_at, r.embedding IS NOT NULL AS has_embedding, ts_rank_cd(r.search_document, websearch_to_tsquery('english', %s)) AS rank FROM project_code_intel_records r """, clauses, """ ORDER BY rank DESC, r.updated_at DESC LIMIT %s """, ) params = [query, *filter_params, query, limit] else: query_sql = query_with_where( """ SELECT r.id, r.snapshot_id, r.collection, r.repo, r.repo_role, r.branch, r.commit_sha, r.tree_sha, r.source_path, r.language, r.file_role, r.content_class, r.record_type, r.record_id, r.parent_record_id, r.title, r.summary, r.line_start, r.line_end, r.symbol, r.symbol_kind, r.confidence_kind, r.confidence, r.tool, r.rule_id, r.severity, r.metadata, r.updated_at, r.embedding IS NOT NULL AS has_embedding, NULL::real AS rank FROM project_code_intel_records r """, clauses, """ ORDER BY r.updated_at DESC LIMIT %s """, ) params = [*filter_params, limit] with db.connect() as conn: if not code_intel_tables_exist(conn): return ok({"error": "code intelligence schema is not initialized"}) rows = conn.execute( db.query_sql(query_sql), params, ).fetchall() return ok({"query": query, **snapshot_scope_response(args), "results": rows}) - Tool definition and input schema for search_code_intel_text. Describes the tool and declares optional parameters: query, limit, collection, repo, record_type, language, file_role, content_class, confidence_kind, source_path, symbol, metadata_key, metadata_value, metadata_contains, snapshot_id, include_historical.
"search_code_intel_text": ToolDefinition( "Search or list code intelligence records with optional PostgreSQL full-text search and exact filters.", { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 50}, "collection": {"type": "string"}, "repo": {"type": "string"}, "record_type": {"type": "string"}, "language": {"type": "string"}, "file_role": {"type": "string"}, "content_class": {"type": "string"}, "confidence_kind": {"type": "string"}, "source_path": {"type": "string"}, "symbol": {"type": "string"}, "metadata_key": {"type": "string"}, "metadata_value": {"type": "string"}, "metadata_contains": {"type": "object"}, "snapshot_id": {"type": "integer", "minimum": 1}, "include_historical": {"type": "boolean"}, }, "additionalProperties": False, }, ), - src/project_code_intelligence/server.py:522-531 (registration)Registration of search_code_intel_text in the TOOLS registry, mapping the tool name to its ToolDefinition and handler function.
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), } - code_intel_clauses helper generates SQL WHERE clauses and parameters for filtering code intelligence records by collection, repo, record_type, language, file_role, content_class, confidence_kind, source_path, symbol, metadata, and snapshot scope.
def code_intel_clauses(args: Json, alias: str = "") -> tuple[list[str], QueryParams]: clauses = ["TRUE"] params: QueryParams = [] collection = scoped_collection(args) if collection: clauses.append(f"{column(alias, 'collection')} = %s") params.append(collection) for name in ("repo", "record_type", "language", "file_role", "content_class", "confidence_kind"): value = optional_text(args, name) if value: clauses.append(f"{column(alias, name)} = %s") params.append(value) source_path = optional_text(args, "source_path") if source_path: clauses.append(f"{column(alias, 'source_path')} = %s") params.append(source_path) symbol = optional_text(args, "symbol") if symbol: clauses.append(f"{column(alias, 'symbol')} = %s") params.append(symbol) metadata_key = optional_text(args, "metadata_key") metadata_value = optional_text(args, "metadata_value") if metadata_key and metadata_value: clauses.append(f"{column(alias, 'metadata')}->>%s = %s") params.extend([metadata_key, metadata_value]) elif metadata_key: clauses.append(f"{column(alias, 'metadata')} ? %s") params.append(metadata_key) metadata_contains = args.get("metadata_contains") if metadata_contains is not None: if not isinstance(metadata_contains, dict): raise McpProtocolTypeError("metadata_contains must be an object") clauses.append(f"{column(alias, 'metadata')} @> %s::jsonb") params.append(json_argument(metadata_contains, "metadata_contains")) snapshot_clauses, snapshot_params = scoped_snapshot_clauses(args, alias) clauses.extend(snapshot_clauses) params.extend(snapshot_params) return clauses, params - optional_text helper extracts and validates an optional string argument from the JSON args. Returns None if absent/empty, raises McpProtocolTypeError if not a string, and checks max text length.
def optional_text(args: Json, name: str) -> str | None: value = args.get(name) if value is None: return None if isinstance(value, str) and not value: return None if not isinstance(value, str): raise McpProtocolTypeError(f"{name} must be a string") if len(value) > mcp_max_text_chars(): raise McpProtocolError(f"{name} exceeds PROJECT_CODE_INTELLIGENCE_MCP_MAX_TEXT_CHARS") return value