Jinni: Bring Your Project Into Context

Instructions

Implementation Reference

• jinni/server.py:48-297 (handler)

  MCP tool handler for 'read_context'. This async function defines the tool logic, input validation using Pydantic Fields (schema), path translation, exclusion handling, and delegates to core_read_context from core_logic.py. Registered via @server.tool decorator.
  @server.tool(description=( "Reads context from a specified project root directory (absolute path). " "Focuses on the specified target files/directories within that root. " "Returns a static view of files with paths relative to the project root. " "Assume the user wants to read in context for the whole project unless otherwise specified - " "do not ask the user for clarification if just asked to read context. " "If the user just says 'jinni', interpret that as read_context. " "If the user asks to list context, use the list_only argument. " "Both `targets` and `rules` accept a JSON array of strings. " "The `project_root`, `targets`, and `rules` arguments are mandatory. " "You can ignore the other arguments by default. " "IMPORTANT NOTE ON RULES: Ensure you understand the rule syntax (details available via the `usage` tool) before providing specific rules. " "Using `rules=[]` is recommended if unsure, as this uses sensible defaults.\n\n" "**Guidance for AI Model Usage**\n\n" "When requesting context using this tool:\n" "* **Default Behavior:** If you provide an empty `rules` list (`[]`), Jinni uses sensible default exclusions (like `.git`, `node_modules`, `__pycache__`, common binary types) combined with any project-specific `.contextfiles`. This usually provides the \"canonical context\" - files developers typically track in version control. Assume this is what the users wants if they just ask to read context.\n" "* **Targeting Specific Files:** If you have a list of specific files you need (e.g., `[\"src/main.py\", \"README.md\"]`), provide them in the `targets` list. This is efficient and precise, quicker than reading one by one.\n" )) async def read_context( project_root: str = Field(description="**MUST BE ABSOLUTE PATH**. The absolute path to the project root directory."), targets: List[str] = Field(description="**Mandatory**. List of paths (absolute or relative to CWD) to specific files or directories within the project root to process. Must be a JSON array of strings. If empty (`[]`), the entire `project_root` is processed."), rules: List[str] = Field(description="**Mandatory**. List of inline filtering rules. Provide `[]` if no specific rules are needed (uses defaults). It is strongly recommended to consult the `usage` tool documentation before providing a non-empty list."), list_only: bool = False, size_limit_mb: Optional[int] = None, debug_explain: bool = False, exclusions: Optional[dict] = Field(default=None, description="Optional exclusion configuration. Object with 'global' (list of keywords), 'scoped' (object mapping paths to keyword lists), and 'patterns' (list of file patterns) fields."), ) -> str: logger.info("--- read_context tool invoked ---") # Translate incoming paths *before* any validation or Path object creation translated_project_root = _translate_wsl_path(project_root) translated_targets = [_translate_wsl_path(t) for t in targets] logger.debug(f"Original paths: project_root='{project_root}', targets='{targets}'") logger.debug(f"Translated paths: project_root='{translated_project_root}', targets='{translated_targets}'") # Defensive NUL check on all incoming paths ensure_no_nul(translated_project_root, "project_root") for t in translated_targets: ensure_no_nul(t, "target path") logger.debug(f"Processing read_context request: project_root(orig)='{project_root}', targets(orig)='{targets}', list_only={list_only}, rules={rules}, debug_explain={debug_explain}") """ Generates a concatenated view of relevant code files for a given target path. The 'project_root' argument must always be an absolute path. The optional 'targets' argument, if provided, must be a list of paths (JSON array of strings). Each path must be absolute or relative to the current working directory, and must resolve to a location *inside* the 'project_root'. If the server was started with a --root argument, the provided 'project_root' must be within that server root directory. Args: project_root: See Field description. targets: See Field description. rules: See Field description. list_only: Only list file paths found. Defaults to False. size_limit_mb: Override the maximum total context size in MB. Defaults to None (uses core_logic default). debug_explain: Print detailed explanation for file/directory inclusion/exclusion to server's stderr. Defaults to False. """ # --- Input Validation --- # Use the translated project_root for validation if not os.path.isabs(translated_project_root): raise ValueError(f"Tool 'project_root' argument must be absolute (after translation), received: '{translated_project_root}' from original '{project_root}'") resolved_project_root_path = Path(translated_project_root).resolve() if not resolved_project_root_path.is_dir(): raise ValueError(f"Tool 'project_root' path does not exist or is not a directory: {resolved_project_root_path} (translated from '{project_root}')") resolved_project_root_path_str = str(resolved_project_root_path) # Store translated path as string logger.debug(f"Using project_root (translated): {resolved_project_root_path_str}") # Validate mandatory targets list (can be empty) # No need for `is None` check, Pydantic/FastMCP ensures it's a list. resolved_target_paths_str: List[str] = [] effective_targets_set: Set[str] = set() # Use set to handle duplicates implicitly # Process the provided targets list if it's not empty if translated_targets: logger.debug(f"Processing translated targets list: {translated_targets}") for idx, single_target in enumerate(translated_targets): if not isinstance(single_target, str): raise TypeError(f"Tool 'targets' item at index {idx} must be a string, got {type(single_target)}") # Check if target is absolute. If not, resolve relative to project_root. target_path_obj = Path(single_target) if target_path_obj.is_absolute(): resolved_target_path = target_path_obj.resolve() else: # Resolve relative path against the project root resolved_target_path = (resolved_project_root_path / target_path_obj).resolve() logger.debug(f"Resolved relative target '{single_target}' to '{resolved_target_path}' using project root '{resolved_project_root_path}'") if not resolved_target_path.exists(): raise FileNotFoundError(f"Tool 'targets' path '{single_target}' (resolved to {resolved_target_path}) does not exist.") # Check if target is within project_root AFTER resolving try: resolved_target_path.relative_to(resolved_project_root_path) except ValueError: raise ValueError(f"Tool 'targets' path '{resolved_target_path}' is outside the specified project root '{resolved_project_root_path}'") resolved_path_str = str(resolved_target_path) if resolved_path_str not in effective_targets_set: resolved_target_paths_str.append(resolved_path_str) effective_targets_set.add(resolved_path_str) logger.debug(f"Validated target path from targets[{idx}]: {resolved_path_str}") else: logger.debug(f"Skipping duplicate target path from targets[{idx}]: {resolved_path_str}") # If the initial targets list was empty OR it resulted in an empty list after validation, # default to processing the project root. if not resolved_target_paths_str: logger.debug("Targets list is empty or resulted in no valid paths. Defaulting to project root.") resolved_target_paths_str = [resolved_project_root_path_str] # Validate mandatory rules list (can be empty, but must be provided) if rules is None: # Should not happen if Pydantic enforces mandatory, but good practice raise ValueError("Tool 'rules' argument is mandatory. Provide an empty list [] if no specific rules are needed.") if not isinstance(rules, list): raise TypeError(f"Tool 'rules' argument must be a list, got {type(rules)}") for idx, rule in enumerate(rules): if not isinstance(rule, str): raise TypeError(f"Tool 'rules' item at index {idx} must be a string, got {type(rule)}") logger.debug(f"Using provided rules: {rules}") # --- Validate against Server Root (if set) --- # The *project_root* provided by the client must be within the server's root (if set) if SERVER_ROOT_PATH: logger.debug(f"Server root is set: {SERVER_ROOT_PATH}") try: resolved_project_root_path.relative_to(SERVER_ROOT_PATH) logger.debug(f"Client project_root {resolved_project_root_path} is within server root {SERVER_ROOT_PATH}") except ValueError: raise ValueError(f"Tool project_root '{resolved_project_root_path}' is outside the allowed server root '{SERVER_ROOT_PATH}'") # --- Process Exclusions --- exclusion_parser = None exclusion_patterns = [] if exclusions: from jinni.exclusion_parser import ExclusionParser # Extract exclusion components global_keywords = exclusions.get('global', []) scoped_exclusions = exclusions.get('scoped', {}) file_patterns = exclusions.get('patterns', []) # Convert scoped dict to list format expected by ExclusionParser scoped_list = [] for path, keywords in scoped_exclusions.items(): if isinstance(keywords, list): scoped_list.append(f"{path}:{','.join(keywords)}") # Create exclusion parser parser = ExclusionParser() # Parse different exclusion types exclusion_patterns.extend(parser.parse_not(global_keywords)) exclusion_patterns.extend(parser.parse_not_in(scoped_list)) exclusion_patterns.extend(parser.parse_not_files(file_patterns)) if exclusion_patterns: exclusion_parser = parser logger.info(f"Configured {len(exclusion_patterns)} exclusion patterns") logger.info(f"Processing project_root: {resolved_project_root_path_str}") # Log the final list of targets being processed # Log the final list of targets being processed # Log the final list of targets being processed logger.info(f"Focusing on target(s): {resolved_target_paths_str}") # --- Call Core Logic --- log_capture_buffer = None temp_handler = None loggers_to_capture = [] debug_output = "" try: if debug_explain: # Setup temporary handler to capture debug logs log_capture_buffer = io.StringIO() temp_handler = logging.StreamHandler(log_capture_buffer) temp_handler.setLevel(logging.DEBUG) # Simple formatter for captured logs formatter = logging.Formatter('%(name)s:%(levelname)s: %(message)s') temp_handler.setFormatter(formatter) # Add handler to relevant core logic loggers loggers_to_capture = [ logging.getLogger(name) for name in ["jinni.core_logic", "jinni.context_walker", "jinni.file_processor", "jinni.config_system", "jinni.utils"] ] for core_logger in loggers_to_capture: # Explicitly set level to DEBUG *before* adding handler # This ensures messages are generated for the handler to capture original_level = core_logger.level core_logger.setLevel(logging.DEBUG) core_logger.addHandler(temp_handler) # Store original level? Not strictly necessary for this hack, # as we remove the handler later, but good practice if we were restoring level. # Pass the validated list of target paths (or the project root if no target was given) # The variable resolved_target_paths_str already holds the correct list. effective_target_paths_str = resolved_target_paths_str # Combine rules with exclusion patterns effective_rules = rules.copy() if rules else [] if exclusion_patterns: effective_rules.extend(exclusion_patterns) # Call the core logic function result_content = core_read_context( target_paths_str=effective_target_paths_str, project_root_str=resolved_project_root_path_str, # Pass the translated, validated root override_rules=effective_rules, list_only=list_only, size_limit_mb=size_limit_mb, debug_explain=debug_explain, # Pass flag down # include_size_in_list is False by default in core_logic if not passed exclusion_parser=exclusion_parser # Pass exclusion parser for scoped exclusions ) logger.debug(f"Finished processing project_root: {resolved_project_root_path_str}, targets(s): {resolved_target_paths_str}. Result length: {len(result_content)}") if debug_explain and log_capture_buffer: debug_output = log_capture_buffer.getvalue() # Combine result and debug output if necessary if debug_output: return f"{result_content}\n\n--- DEBUG LOG ---\n{debug_output}" else: return result_content except (FileNotFoundError, ContextSizeExceededError, ValueError, DetailedContextSizeError) as e: # Let FastMCP handle converting these known errors logger.error(f"Error during read_context call for project_root='{resolved_project_root_path_str}', targets(s)='{resolved_target_paths_str}': {type(e).__name__} - {e}") raise e # Re-raise for FastMCP except Exception as e: # Log unexpected errors before FastMCP potentially converts to a generic 500 logger.exception(f"Unexpected error processing project_root='{resolved_project_root_path_str}', targets(s)='{resolved_target_paths_str}': {type(e).__name__} - {e}") raise e finally: # --- Cleanup: Remove temporary handler --- if temp_handler and loggers_to_capture: logger.debug("Removing temporary debug log handler.") for core_logger in loggers_to_capture: core_logger.removeHandler(temp_handler) temp_handler.close()
• jinni/core_logic.py:44-293 (helper)

  Core implementation logic delegated by the MCP handler. Orchestrates context processing: input validation, root determination, size limits, override rules, directory walking, file processing, and output formatting.
  def read_context( target_paths_str: List[str], # List of targets from CLI or constructed by Server project_root_str: Optional[str] = None, # Optional from CLI, Mandatory from Server (used as base) override_rules: Optional[List[str]] = None, list_only: bool = False, size_limit_mb: Optional[int] = None, debug_explain: bool = False, include_size_in_list: bool = False, exclusion_parser: Optional[Any] = None # ExclusionParser instance for scoped exclusions ) -> str: """ Orchestrates the context reading process, handling flexible inputs. Validates inputs, determines the effective roots for rule discovery and output, resolves targets, and delegates processing to file_processor or context_walker. Args: target_paths_str: List of target file/directory paths (relative or absolute). project_root_str: Optional path to the project root. If provided, it's used as the base for rule discovery and output relativity. If None, it's inferred from the common ancestor of targets. override_rules: Optional list of rule strings to use instead of .contextfiles. list_only: If True, only return a list of relative file paths. size_limit_mb: Optional override for the size limit in MB. debug_explain: If True, log inclusion/exclusion reasons. include_size_in_list: If True and list_only, prepend size to path. Returns: A formatted string (concatenated content or file list). Raises: FileNotFoundError: If any target path does not exist. ValueError: If paths have issues (e.g., target outside explicit root). DetailedContextSizeError: If context size limit is exceeded. ImportError: If pathspec is required but not installed. """ # --- Initial Setup & Validation --- # Validate project_root_str FIRST if provided, and set roots output_rel_root: Path rule_discovery_root: Path project_root_path: Optional[Path] = None # Store resolved explicit project_root if project_root_str: project_root_path = Path(project_root_str).resolve() if not project_root_path.is_dir(): # Raise ValueError immediately if explicit root is invalid raise ValueError(f"Provided project root '{project_root_str}' does not exist or is not a directory.") output_rel_root = project_root_path rule_discovery_root = project_root_path logger.debug(f"Using provided project root for output relativity and rule discovery boundary: {output_rel_root}") # else: Roots will be determined after resolving targets # Resolve target paths (relative to CWD by default) target_paths: List[Path] = [] if not target_paths_str: # Handle case where CLI provides no paths (defaults to ['.']) # or Server provides no target (meaning process root) if project_root_str: # If root is given but no targets, process the root target_paths_str = [project_root_str] logger.debug("No specific targets provided; processing project root.") else: # If no root and no targets, default to current dir '.' target_paths_str = ['.'] logger.debug("No specific targets or project root provided; processing current directory '.'") for p_str in target_paths_str: p = Path(p_str).resolve() # Resolve paths here to ensure they are absolute if not p.exists(): raise FileNotFoundError(f"Target path does not exist: {p_str} (resolved to {p})") target_paths.append(p) if not target_paths: logger.warning("No valid target paths could be determined.") return "" # Determine roots IF project_root wasn't provided explicitly if not project_root_path: try: common_ancestor = Path(os.path.commonpath([str(p) for p in target_paths])) calculated_root = common_ancestor if common_ancestor.is_dir() else common_ancestor.parent except ValueError: logger.warning("Could not find common ancestor for targets. Using CWD as root.") calculated_root = Path.cwd().resolve() output_rel_root = calculated_root rule_discovery_root = calculated_root logger.debug(f"Using common ancestor/CWD as output relativity and rule discovery boundary root: {output_rel_root}") # else: Roots were already set from the valid project_root_path # Ensure roots are set (safeguard) if 'output_rel_root' not in locals() or 'rule_discovery_root' not in locals(): logger.error("Critical error: Output/Rule discovery root could not be determined.") raise ValueError("Could not determine a root directory.") # Validate targets are within explicit root (if provided) AFTER resolving roots if project_root_path: for tp in target_paths: try: # Use is_relative_to for Python 3.9+ or fallback if sys.version_info >= (3, 9): if not tp.is_relative_to(project_root_path): raise ValueError(f"Target path {tp} is outside the specified project root {project_root_path}") else: tp.relative_to(project_root_path) # Check raises ValueError if not relative except ValueError: raise ValueError(f"Target path {tp} is outside the specified project root {project_root_path}") # (Logic moved above) # --- Size Limit (Moved up slightly, no functional change) --- limit_mb_str = os.environ.get(ENV_VAR_SIZE_LIMIT) try: effective_limit_mb = size_limit_mb if size_limit_mb is not None \ else int(limit_mb_str) if limit_mb_str else DEFAULT_SIZE_LIMIT_MB except ValueError: logger.warning(f"Invalid value for {ENV_VAR_SIZE_LIMIT} ('{limit_mb_str}'). Using default {DEFAULT_SIZE_LIMIT_MB}MB.") effective_limit_mb = DEFAULT_SIZE_LIMIT_MB size_limit_bytes = effective_limit_mb * 1024 * 1024 logger.debug(f"Effective size limit: {effective_limit_mb}MB ({size_limit_bytes} bytes)") # --- Override Handling --- # Override rules are used only if the list is provided AND non-empty use_overrides = bool(override_rules) # bool([]) is False, bool(['rule']) is True override_spec: Optional['pathspec.PathSpec'] = None if use_overrides: if pathspec is None: raise ImportError("pathspec library is required for override rules but not installed.") logger.info("Override rules provided as high-priority additions to normal rules.") # Store the override rules for later use in the walker override_spec = compile_spec_from_rules(override_rules, "Overrides") if debug_explain: logger.debug(f"Compiled override spec with {len(override_spec.patterns)} patterns.") # --- Processing State --- output_parts: List[str] = [] processed_files_set: Set[Path] = set() total_size_bytes: int = 0 # Use the resolved target_paths as the initial set for "always include" logic within walker/processor initial_target_paths_set: Set[Path] = set(target_paths) # --- Compute a root for each target --- roots_for_target: Dict[Path, Path] = {} # Determine the base for comparison (project root or CWD) comparison_base = project_root_path if project_root_path else Path.cwd().resolve() for tp in target_paths: try: # Check if target is within the comparison base if tp.is_relative_to(comparison_base): root = comparison_base # Use project root (or CWD) for targets within it else: root = tp if tp.is_dir() else tp.parent # External targets are self-contained except AttributeError: # Python < 3.9 fallback if str(tp).startswith(str(comparison_base)): root = comparison_base else: root = tp if tp.is_dir() else tp.parent roots_for_target[tp] = root if debug_explain: logger.debug(f"Target {tp} will use rule root: {root}") # --- Delegate Processing --- try: # Group targets by their root and walk once per root for root in set(roots_for_target.values()): # Collect targets that belong to this root grouped = [t for t, r in roots_for_target.items() if r == root] # Build an "always include" set for this root initial_set = set(grouped) for current_target_path in grouped: # Skip if already processed (e.g., listed twice or handled by a previous dir walk) if current_target_path in processed_files_set: if debug_explain: logger.debug(f"Skipping target {current_target_path} as it was already processed.") continue if current_target_path.is_file(): if debug_explain: logger.debug(f"Processing file target: {current_target_path}") file_output, file_size_added = process_file( file_path=current_target_path, output_rel_root=output_rel_root, # Use the original output root size_limit_bytes=size_limit_bytes, total_size_bytes=total_size_bytes, list_only=list_only, include_size_in_list=include_size_in_list, debug_explain=debug_explain ) if file_output is not None: # Check if adding this file *content* exceeds limit (only if not list_only) if not list_only and (total_size_bytes + file_size_added > size_limit_bytes): # Check if file alone exceeds limit if file_size_added > size_limit_bytes and total_size_bytes == 0: logger.warning(f"File {current_target_path} ({file_size_added} bytes) content exceeds size limit of {effective_limit_mb}MB. Skipping.") continue # Skip this file else: # Adding this file pushes over the limit raise ContextSizeExceededError(effective_limit_mb, total_size_bytes + file_size_added, current_target_path) output_parts.append(file_output) processed_files_set.add(current_target_path) total_size_bytes += file_size_added # Add size only if content included elif current_target_path.is_dir(): if debug_explain: logger.debug(f"Processing directory target: {current_target_path}") dir_output_parts, dir_total_size, dir_processed_files = walk_and_process( walk_target_path=current_target_path, rule_root=root, # Pass the rule root for this target output_rel_root=output_rel_root, # Keep original output root for consistent paths initial_target_paths_set=initial_set, # Pass initial targets for this root use_overrides=use_overrides, override_spec=override_spec, size_limit_bytes=size_limit_bytes - total_size_bytes, # Pass remaining budget list_only=list_only, include_size_in_list=include_size_in_list, debug_explain=debug_explain, exclusion_parser=exclusion_parser # Pass exclusion parser for scoped exclusions ) output_parts.extend(dir_output_parts) processed_files_set.update(dir_processed_files) total_size_bytes += dir_total_size # Accumulate size from walker else: logger.warning(f"Target path is neither a file nor a directory: {current_target_path}") # --- Final Output Formatting --- final_output = "\n".join(output_parts) if list_only else SEPARATOR.join(output_parts) logger.info(f"Processed {len(processed_files_set)} files, total size: {total_size_bytes} bytes.") return final_output except ContextSizeExceededError as e: # Catch error, format with large files list, and raise Detailed error logger.error(f"Context size limit exceeded: {e}") # Use output_rel_root as the base for finding large files large_files = get_large_files(str(output_rel_root)) error_message = ( f"Error: Context size limit of {e.limit_mb}MB exceeded.\n" f"Processing stopped near file: {e.file_path}\n\n" "Consider excluding large files or directories using a `.contextfiles` file.\n" "Consult README.md (or use `jinni doc`) for more details on exclusion rules.\n\n" "Potential large files found (relative to project root):\n" ) if large_files: for fname, fsize in large_files: size_mb = fsize / (1024 * 1024) error_message += f" - {fname} ({size_mb:.2f} MB)\n" else: error_message += " - Could not identify specific large files.\n" raise DetailedContextSizeError(error_message) from e