CrewAI + Nexus Integration¶

from crewai.tools import tool

@tool("Read File")
def read_file(path: str) -> str:
    """Read a file from Nexus filesystem.

    Args:
        path: File path to read (e.g., '/workspace/data.txt')

    Returns:
        File content as string
    """
    return call_nexus_mcp("nexus_read_file", path=path)

@tool("Write File")
def write_file(path: str, content: str) -> str:
    """Write content to a file in Nexus filesystem."""
    return call_nexus_mcp("nexus_write_file", path=path, content=content)

@tool("List Files")
def list_files(path: str = "/", recursive: bool = False) -> str:
    """List files in a directory."""
    return call_nexus_mcp("nexus_list_files", path=path, recursive=recursive)

@tool("Search Files by Pattern")
def glob_files(pattern: str, path: str = "/") -> str:
    """Find files matching a glob pattern.

    Examples:
        - glob_files("*.py", "/workspace")  # Find all Python files
        - glob_files("**/*.md", "/docs")    # Find all markdown recursively
    """
    return call_nexus_mcp("nexus_glob", pattern=pattern, path=path)

@tool("Search File Contents")
def grep_files(pattern: str, path: str = "/") -> str:
    """Search file contents using regex pattern.

    Examples:
        - grep_files("async def", "/workspace")  # Find async functions
        - grep_files("TODO:", "/")               # Find TODO comments
    """
    return call_nexus_mcp("nexus_grep", pattern=pattern, path=path)

@tool("Semantic Search")
def semantic_search(query: str, limit: int = 10) -> str:
    """Search files semantically using natural language.

    Examples:
        - semantic_search("authentication setup")
        - semantic_search("error handling patterns")
    """
    return call_nexus_mcp("nexus_semantic_search", query=query, limit=limit)

@tool("Store Memory")
def store_memory(content: str, memory_type: str = None, importance: float = 0.5) -> str:
    """Store a memory in Nexus for long-term learning.

    Args:
        content: Memory content to store
        memory_type: Optional category (e.g., 'insight', 'fact')
        importance: Importance score 0.0-1.0 (default: 0.5)

    Examples:
        - store_memory("Always validate input", "security_insight", 0.9)
        - store_memory("Prefer async for I/O operations", "pattern", 0.7)
    """
    return call_nexus_mcp("nexus_store_memory",
                         content=content,
                         memory_type=memory_type,
                         importance=importance)

@tool("Query Memory")
def query_memory(query: str, limit: int = 5) -> str:
    """Retrieve relevant memories using semantic search.

    Examples:
        - query_memory("security best practices")
        - query_memory("performance optimization tips")
    """
    return call_nexus_mcp("nexus_query_memory", query=query, limit=limit)

from crewai import Agent, Task, Crew

# Create agent with file operation tools
analyst = Agent(
    role="Code Analyst",
    goal="Analyze Python files and identify patterns",
    backstory="""You are an expert code analyst who excels at finding
    patterns, identifying issues, and providing insights about codebases.""",
    tools=[glob_files, read_file, grep_files, write_file],
    verbose=True,
)

# Create task
task = Task(
    description="""Analyze Python files in /workspace:
    1. Find all Python files using glob
    2. Search for 'async def' patterns using grep
    3. Read a few example files to understand the patterns
    4. Create a summary report at /reports/async-analysis.md""",
    expected_output="A markdown report summarizing async patterns",
    agent=analyst,
)

# Execute
crew = Crew(agents=[analyst], tasks=[task], verbose=True)
result = crew.kickoff()

# Create researcher with memory tools
researcher = Agent(
    role="Research Analyst",
    goal="Research topics and build knowledge over time",
    backstory="""You are a diligent researcher who not only finds
    information but also remembers key insights for future reference.""",
    tools=[semantic_search, read_file, store_memory, query_memory],
    verbose=True,
)

# Task 1: Research and store insights
research_task = Task(
    description="""Research error handling patterns:
    1. Use semantic search to find files about error handling
    2. Read the most relevant files
    3. Store key insights in memory with high importance
    4. Tag memories with type 'error_handling_pattern'""",
    expected_output="Summary of findings with insights stored",
    agent=researcher,
)

# Task 2: Recall and synthesize
synthesis_task = Task(
    description="""Synthesize previous learnings:
    1. Query memory for 'error_handling_pattern' insights
    2. Combine with any new findings
    3. Create a comprehensive guide""",
    expected_output="Guide based on stored memories",
    agent=researcher,
)

# Execute both tasks
crew = Crew(
    agents=[researcher],
    tasks=[research_task, synthesis_task],
    verbose=True
)
result = crew.kickoff()

# Agent 1: Data Collector
collector = Agent(
    role="Data Collector",
    goal="Collect and organize information from files",
    backstory="""You are a meticulous data collector who finds and
    organizes information systematically.""",
    tools=[glob_files, grep_files, list_files, write_file],
    verbose=True,
)

# Agent 2: Senior Analyst
analyst = Agent(
    role="Senior Analyst",
    goal="Analyze collected data and generate insights",
    backstory="""You are an experienced analyst who reads collected
    data, identifies patterns, and generates actionable insights.""",
    tools=[read_file, semantic_search, write_file, store_memory],
    verbose=True,
)

# Task 1: Collect TODOs
collect_task = Task(
    description="""Collect TODO comments from the codebase:
    1. Search for 'TODO' patterns in all files
    2. Organize findings by file
    3. Write collected data to /workspace/todo-collection.txt""",
    expected_output="Organized list of TODO items saved to file",
    agent=collector,
)

# Task 2: Analyze TODOs (depends on collect_task)
analyze_task = Task(
    description="""Analyze the collected TODO items:
    1. Read /workspace/todo-collection.txt
    2. Categorize TODOs by priority and type
    3. Generate recommendations
    4. Write report to /reports/todo-analysis.md
    5. Store key insights in memory""",
    expected_output="Analysis report with recommendations",
    agent=analyst,
)

# Execute sequential workflow
crew = Crew(
    agents=[collector, analyst],
    tasks=[collect_task, analyze_task],
    verbose=True,
)
result = crew.kickoff()

# Store insight
nx.memory.store(
    "Always use parameterized queries to prevent SQL injection",
    scope="user",
    memory_type="security_pattern",
    importance=0.95
)

# Later... retrieve relevant memories
memories = nx.memory.search(
    query="database security best practices",
    limit=5
)

# Terminal 1: Nexus server
./start_nexus_server.sh

# Terminal 2: CrewAI agents connect remotely
nx = nexus.connect(config={
    "remote_url": "http://localhost:2026",
    "api_key": os.getenv("NEXUS_API_KEY")
})