crewai

CrewAI Integration Guide

Use mcp-memory-service as the persistent shared memory backend for CrewAI agents and crews.

Setup

pip install mcp-memory-service crewai httpx
MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http

Custom Memory Tools

Implement BaseTool subclasses to expose memory as CrewAI tools:

import httpx
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

MEMORY_URL = "http://localhost:8000"


class SearchMemoryInput(BaseModel):
    query: str = Field(description="Natural language search query")
    tags: list[str] = Field(default=[], description="Optional tag filters (e.g. ['agent:researcher'])")
    limit: int = Field(default=5, description="Maximum number of results")


class MemorySearchTool(BaseTool):
    name: str = "Search Memory"
    description: str = (
        "Search long-term shared memory for relevant context. "
        "Use tags like 'agent:researcher' to scope results to a specific agent."
    )
    args_schema: type[BaseModel] = SearchMemoryInput

    def _run(self, query: str, tags: list[str] = None, limit: int = 5) -> str:
        import asyncio
        return asyncio.run(self._arun(query, tags or [], limit))

    async def _arun(self, query: str, tags: list[str] = None, limit: int = 5) -> str:
        payload = {"query": query, "limit": limit}
        if tags:
            payload["tags"] = tags

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{MEMORY_URL}/api/memories/search",
                json=payload,
            )
            memories = response.json().get("memories", [])

        if not memories:
            return "No relevant memories found."
        return "\n".join(f"[{', '.join(m['tags'])}] {m['content']}" for m in memories)


class StoreMemoryInput(BaseModel):
    content: str = Field(description="Memory content to store")
    tags: list[str] = Field(default=[], description="Tags to categorize the memory")
    memory_type: str = Field(default="note", description="Memory type: note, observation, decision, fact")


class MemoryStoreTool(BaseTool):
    name: str = "Store Memory"
    description: str = (
        "Store an important finding, decision, or fact in long-term shared memory. "
        "Other agents in the crew can retrieve it later."
    )
    args_schema: type[BaseModel] = StoreMemoryInput
    agent_id: str = ""  # Set when creating tool instance

    def _run(self, content: str, tags: list[str] = None, memory_type: str = "note") -> str:
        import asyncio
        return asyncio.run(self._arun(content, tags or [], memory_type))

    async def _arun(self, content: str, tags: list[str] = None, memory_type: str = "note") -> str:
        headers = {"Content-Type": "application/json"}
        if self.agent_id:
            headers["X-Agent-ID"] = self.agent_id

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{MEMORY_URL}/api/memories",
                json={"content": content, "tags": tags or [], "memory_type": memory_type},
                headers=headers,
            )
            result = response.json()

        if result.get("success"):
            return f"Stored memory (hash: {result['content_hash']})"
        return f"Failed to store memory: {result.get('message', 'unknown error')}"