Share Memory Context Across CrewAI Agent Crews

import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from sovseal import AgentStateClient, EngineUnavailableError

load_dotenv()

# store/recall need no endpoint/api_key; they talk to the local native host.
client = AgentStateClient()


@tool("recall_memory")
def recall_memory(query: str) -> str:
    """Search the local memory store for facts relevant to a query."""
    try:
        hits = client.recall(query, top_k=5)
    except EngineUnavailableError as e:
        return f"Memory engine unavailable ({e.reason})."
    if not hits:
        return "No relevant memories found."
    return "\n".join(f"- {h['text']}" for h in hits)


@tool("store_memory")
def store_memory(content: str) -> str:
    """Persist a new fact or constraint to the local memory store."""
    try:
        client.store(content)
    except EngineUnavailableError as e:
        return f"Memory engine unavailable ({e.reason}) — not stored."
    return "Stored."


def run_crew():
    memory_tools = [recall_memory, store_memory]

    # Researcher agent — discovers and stores constraints
    researcher = Agent(
        role="Lead Project Researcher",
        goal="Identify and index key architecture constraints for the current project",
        backstory=(
            "You are an expert researcher. Before starting any analysis, "
            "use recall_memory to check what constraints have been found in "
            "previous runs. After your analysis, use store_memory to persist "
            "new findings."
        ),
        tools=memory_tools,
        verbose=True,
    )

    # Writer agent — drafts output informed by recalled constraints
    writer = Agent(
        role="Technical Writer",
        goal="Produce accurate architecture documentation",
        backstory=(
            "You are a precise technical writer. Always call recall_memory "
            "at the start of your task to retrieve architecture constraints "
            "indexed by the researcher."
        ),
        tools=memory_tools,
        verbose=True,
    )

    # Tasks
    research_task = Task(
        description=(
            "Query the local memory node for 'database choices' and "
            "'API design constraints'. Summarise all findings and store "
            "any new constraints you identify."
        ),
        expected_output=(
            "A structured list of database technologies and API constraints "
            "verified across all previous crew runs."
        ),
        agent=researcher,
    )

    writing_task = Task(
        description=(
            "Using the constraints recalled from memory, draft a concise "
            "architecture decision record (ADR) for the project."
        ),
        expected_output="A markdown ADR document covering all recalled constraints.",
        agent=writer,
        context=[research_task],
    )

    # Assemble and run the crew
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, writing_task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    print("Crew Result:", result)


if __name__ == "__main__":
    run_crew()