File Operations Toolkit

pip install marsys  # Includes PyMuPDF and Pillow

pip install tree-sitter

import os
from pathlib import Path
from marsys import Agent
from marsys.models import ModelConfig
from marsys.environment import FileOperationConfig, create_file_operation_tools
from marsys.environment.filesystem import RunFileSystem

# Create model configuration (example using OpenRouter)
# Note: API key is only required if you use API-based models
model_config = ModelConfig(
    type="api",
    name="anthropic/claude-opus-4.6",
    provider="openrouter",
    api_key=os.getenv("OPENROUTER_API_KEY")  # Set if using OpenRouter
)

# Create file operation tools
# Shared run filesystem for all agents
fs = RunFileSystem.local(run_root=Path("./runs/run-20260206"))
file_config = FileOperationConfig(run_filesystem=fs)
file_tools = create_file_operation_tools(file_config)

# Create agent with file capabilities
file_agent = Agent(
    model_config=model_config,
    goal="Manage and analyze files efficiently",
    instruction="""You are a file management assistant. Use your file operation tools to:
    - Read files intelligently based on size and type
    - Extract structured information from documents
    - Edit files using unified diff format for reliability
    - Search for content across multiple files
    - Maintain security by respecting run filesystem boundaries

    Always use the most appropriate reading strategy to optimize token usage.
    When editing, prefer unified diff format for complex changes.""",
    name="FileAssistant",
    tools=file_tools
)

# Use the agent
result = await file_agent.run(
    prompt="Read the README.md file and summarize its contents"
)

from pathlib import Path
from marsys.environment.filesystem import RunFileSystem
from marsys.environment import FileOperationConfig, create_file_operation_tools

fs = RunFileSystem.local(run_root=Path("./runs/run-20260206"))

config = FileOperationConfig(run_filesystem=fs)
file_tools = create_file_operation_tools(config)

from pathlib import Path
from marsys.environment import create_file_operation_tools, FileOperationConfig

# Use defaults (permissive mode)
file_tools = create_file_operation_tools()

from pathlib import Path
from marsys.environment import FileOperationConfig, create_file_operation_tools
from marsys.environment.filesystem import RunFileSystem

# Create custom configuration (virtual filesystem + security)
fs = RunFileSystem.local(
    run_root=Path("/home/user/projects"),
    extra_mounts={"/datasets": Path("/shared/datasets")}
)

config = FileOperationConfig(
    base_directory=Path("/home/user/projects"),
    run_filesystem=fs,

    # File size limits (hard limit for safety)
    max_file_size_bytes=100 * 1024 * 1024,  # 100 MB absolute limit

    # Character-based reading thresholds (token proxy for text)
    small_file_threshold=10000,      # < 10k chars (~2.5k tokens): FULL read
    medium_file_threshold=100000,    # 10-100k chars (~25k tokens): PARTIAL read
    large_file_threshold=500000,     # > 500k chars (~125k tokens): OVERVIEW first

    # File type-specific limits
    max_json_content_chars=40000,     # JSON truncation threshold (~10k tokens)
    max_lines_per_read=250,           # Max lines for text files
    max_pages_per_read=5,             # Max pages for PDF files

    # Absolute safety limit (applies to ALL file types)
    max_characters_absolute=120000,   # Hard limit: 120K chars (~30k tokens)

    # Image token limits (for vision models)
    max_image_pixels=1024 * 1024,    # 1 megapixel (1024x1024)
    max_images_per_read=4,           # Maximum images per operation

    # Security patterns (glob-style)
    blocked_patterns=[
        "*.key", "*.pem", "*.p12",  # Private keys
        ".env", ".env.*",             # Environment files
        "*.sqlite", "*.db",           # Databases
        ".git/**",                    # Git internals
    ],

    # Auto-approve patterns (no user confirmation needed)
    auto_approve_patterns=[
        "*.md", "*.txt",              # Documentation
        "*.py", "*.js", "*.java",     # Source code
        "*.json", "*.yaml", "*.yml",  # Configuration
    ],

    # Require approval patterns
    require_approval_patterns=[
        "*.sh", "*.bash",             # Shell scripts
        "Makefile", "Dockerfile",     # Build files
        "*.sql",                      # SQL files
    ],

    # Feature flags
    enable_delete=False,
    enable_tree_sitter=True,
    enable_semantic_search=False,
    enable_caching=True,
    cache_ttl_seconds=300,

    # Search
    max_search_results=100,

    # Audit logging
    enable_audit_logging=True,
    log_file_path=Path("./file_operations_audit.log"),
)

# Create tools with custom config
file_tools = create_file_operation_tools(config)

from marsys.environment import FileOperationConfig, create_file_operation_tools
from pathlib import Path

# Permissive mode (default)
permissive_config = FileOperationConfig.create_permissive()
permissive_tools = create_file_operation_tools(permissive_config)

# Restrictive mode (tighter security)
restrictive_config = FileOperationConfig.create_restrictive(
    base_directory=Path("/workspace")
)
restrictive_tools = create_file_operation_tools(restrictive_config)

from marsys.environment.file_operations import ReadStrategy

# Use FULL strategy for small files
result = await file_agent.run(
    prompt="Read config.yaml using FULL strategy",
    context={"read_strategy": ReadStrategy.FULL}
)

# Use OVERVIEW for large documents
result = await file_agent.run(
    prompt="Get overview of large_report.pdf",
    context={"read_strategy": ReadStrategy.OVERVIEW}
)

# PROGRESSIVE for incremental exploration
result1 = await file_agent.run(
    prompt="Get structure of codebase/main.py",
    context={"read_strategy": ReadStrategy.PROGRESSIVE}
)
result2 = await file_agent.run(
    prompt="Read section 'class:DatabaseManager' from main.py",
    context={"section_id": "class:DatabaseManager"}
)

# Read pages 5-10 of a PDF
result = await file_agent.run(
    prompt="Read pages 5 to 10 from research_paper.pdf",
    context={
        "start_page": 5,
        "end_page": 10
    }
)

# Features:
# - Automatic limit enforcement (prevents requesting too many pages)
# - Clean response without usage guides
# - Returns pure content from requested range

# Read lines 100-200 from a code file
result = await file_agent.run(
    prompt="Read lines 100 to 200 from main.py",
    context={
        "start_line": 100,
        "end_line": 200
    }
)

# Features:
# - Character-based limit enforcement
# - Maximum characters enforced by max_characters_absolute (120K default)
# - Clean response without usage guides (explicit request)

# Request too many pages - returns error response:
# {
#   "error": true,
#   "message": "Request exceeds maximum pages per read",
#   "details": {
#     "requested_pages": 200,
#     "maximum_pages": 100,
#     "suggestion": "Request fewer pages (e.g., start_page=1, end_page=100)"
#   }
# }

# Limits enforced:
# - PDF pages: max_pages_per_read (default: 5 pages)
# - Text lines: max_lines_per_read (default: 250 lines)
# - All file types: max_characters_absolute (default: 120K characters)

# Search for keywords in PDF with page numbers
result = await file_agent.run(
    prompt="Search for 'machine learning' in research_paper.pdf",
    context={
        "search_type": "content",
        "pattern": "machine learning",
        "include_context": True
    }
)

# Result includes page numbers and line numbers:
# {
#   "matches": [
#     {
#       "match": "Machine learning algorithms...",
#       "location": "page 3, line 45",
#       "page": 3,
#       "line": 45,
#       "context_before": [...],
#       "context_after": [...]
#     }
#   ]
# }

# Read specific section by ID
result = await file_agent.run(
    prompt="Read section '1.2' from research_paper.pdf",
    context={"section_id": "1.2"}
)

result = await file_agent.run(
    prompt="""Edit config.py using unified diff format:

    --- config.py
    +++ config.py
    @@ -10,3 +10,3 @@
     DEBUG = True
    -MAX_WORKERS = 4
    +MAX_WORKERS = 8
     LOG_LEVEL = "INFO"
    """,
    context={}
)

# Features:
# - Multiple fallback strategies (exact match -> whitespace normalization -> fuzzy matching)
# - Dry-run preview before applying
# - Detailed change reports
# - ~98% success rate

result = await file_agent.run(
    prompt="""Replace 'old_function()' with 'new_function()' in utils.py""",
    context={}
)

# Preview changes before applying
result = await file_agent.run(
    prompt="""Show me what would change if I apply this diff (dry run):

    --- app.py
    +++ app.py
    @@ -5,1 +5,1 @@
    -version = "1.0.0"
    +version = "1.1.0"
    """,
    context={"dry_run": True}
)

# Read an image file with token estimation
result = await file_agent.run(
    prompt="Read the diagram.png image",
    context={
        "provider": "anthropic",  # Options: openai, anthropic, google, xai, generic
        "detail": "high",          # Options: high, low (affects some providers)
        "max_pixels": 1024 * 1024  # Downsample if exceeds this limit
    }
)

# Result includes:
# - Image dimensions and format
# - Estimated token count for the provider
# - Base64-encoded image data (for sending to VLM)
# - Metadata (DPI, color mode, etc.)

config = FileOperationConfig(
    # Image pixel limits
    max_image_pixels=1024 * 1024,  # 1 megapixel max per image
    max_images_per_read=4,          # Max images in single operation

    # Auto-downsample if needed
    # Images exceeding max_pixels will be resized maintaining aspect ratio
)

# Read PDF with images (future feature)
result = await file_agent.run(
    prompt="Read report.pdf and include any charts or diagrams",
    context={
        "extract_images": True,
        "provider": "openai",  # For token estimation
        "max_images": 4        # Limit to key visuals
    }
)

# Result includes both text and ImageData objects
print(f"Text tokens: {result.estimated_tokens}")
print(f"Image tokens: {result.total_estimated_image_tokens}")
print(f"Total tokens: {result.get_total_estimated_tokens()}")

from marsys.environment.file_operations.token_estimation import (
    estimate_total_tokens,
    should_downsample_image
)

# Estimate total tokens before reading
estimation = estimate_total_tokens(
    text_content="Sample text...",
    images=[(1920, 1080), (1024, 768)],  # Image dimensions
    provider="anthropic"
)
print(f"Total estimated tokens: {estimation['total_tokens']}")
print(f"  Text: {estimation['text_tokens']} tokens")
print(f"  Images: {estimation['image_tokens']} tokens")

# Check if image needs downsampling
needs_downsample, target_dims = should_downsample_image(
    width=2048,
    height=2048,
    max_pixels=1024 * 1024,
    max_tokens=1000,
    provider="openai"
)
if needs_downsample:
    print(f"Downsample to: {target_dims}")

# Content Search with regex
result = await file_agent.run(
    prompt="Search for 'TODO' comments in all Python files",
    context={
        "search_type": "content",
        "pattern": r"#\s*TODO:",
        "file_pattern": "*.py"
    }
)

# PDF Content Search with page numbers
result = await file_agent.run(
    prompt="Search for 'neural network' in all PDF files",
    context={
        "search_type": "content",
        "pattern": "neural network",
        "file_pattern": "*.pdf",
        "include_context": True,
        "context_lines": 2
    }
)
# Returns matches with page numbers:
# {"matches": [{"match": "...", "location": "page 5, line 23", "page": 5, "line": 23}]}

# Filename Search
result = await file_agent.run(
    prompt="Find all test files",
    context={
        "search_type": "filename",
        "pattern": "**/test_*.py"
    }
)

# Structure Search
result = await file_agent.run(
    prompt="Find all class definitions in the codebase",
    context={
        "search_type": "structure",
        "query": "class:*"
    }
)

from pathlib import Path
from marsys.environment import FileOperationConfig, create_file_operation_tools

config = FileOperationConfig(
    base_directory=Path("/home/user/safe_workspace"),
    extra_mounts={"/datasets": Path("/shared/datasets")}
)

file_tools = create_file_operation_tools(config)

# Attempts to escape the run root are blocked
# Virtual paths like /datasets/* map to the mounted host directory

config = FileOperationConfig(
    # Block these patterns entirely
    blocked_patterns=[
        "*.key", "*.pem",  # Private keys
        ".env*",           # Environment files
        ".git/**",         # Git internals
    ],

    # Auto-approve these patterns (no confirmation)
    auto_approve_patterns=[
        "*.md", "*.txt",   # Safe documents
        "*.json",          # Configuration
    ],

    # Require user approval for these
    require_approval_patterns=[
        "*.sh",            # Shell scripts
        "*.sql",           # Database queries
    ]
)

config = FileOperationConfig(
    max_file_size_bytes=10 * 1024 * 1024,  # 10 MB limit
    max_characters_absolute=80000,          # Absolute read limit (chars)
    max_lines_per_read=200                  # Line-based partial reads
)

config = FileOperationConfig(
    enable_audit_logging=True,
    log_file_path=Path("./file_ops_audit.log")
)

# All operations logged with:
# - Timestamp
# - Operation type
# - File path
# - Success/failure
# - Agent name