http-generic.md

HTTP Generic Integration Guide

Connect any agent framework or HTTP client to mcp-memory-service using the REST API.

Server Setup

pip install mcp-memory-service
MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http
# Running at http://localhost:8000

With API key authentication:

MCP_API_KEY=your-secret-key memory server --http
# Include header: Authorization: Bearer your-secret-key

All REST Endpoints

Method	Endpoint	Description
POST	/api/memories	Store a new memory
GET	/api/memories	List memories (paginated)
GET	/api/memories/{hash}	Get memory by hash
PATCH	/api/memories/{hash}	Update memory tags/type/metadata
DELETE	/api/memories/{hash}	Delete a memory
POST	/api/memories/search	Semantic + keyword search
GET	/api/memories/tags	List all tags with counts
GET	/api/health	Health check
GET	/api/stats	Storage statistics
POST	/api/consolidation/run	Trigger memory consolidation
GET	/api/consolidation/status	Consolidation status
GET	/api/graph/associations/{hash}	Get memory associations
POST	/api/graph/associations	Store association between memories
GET	/api/analytics/relationship-types	Graph relationship type stats
GET	/sse/events	Server-Sent Events stream

Authentication Patterns

# Anonymous (MCP_ALLOW_ANONYMOUS_ACCESS=true)
curl http://localhost:8000/api/memories

# API key
curl -H "Authorization: Bearer $MCP_API_KEY" http://localhost:8000/api/memories

# OAuth (see docs/oauth/)
curl -H "Authorization: Bearer $ACCESS_TOKEN" http://localhost:8000/api/memories

Python (httpx) Examples

Store a memory

import httpx

BASE_URL = "http://localhost:8000"

async def store_memory(content: str, tags: list[str], agent_id: str | None = None) -> dict:
    headers = {"Content-Type": "application/json"}
    if agent_id:
        headers["X-Agent-ID"] = agent_id

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/api/memories",
            json={"content": content, "tags": tags},
            headers=headers,
        )
        response.raise_for_status()
        return response.json()

# Usage
result = await store_memory(
    content="API rate limit is 100 req/min for the Acme service",
    tags=["api", "rate-limit", "acme"],
    agent_id="researcher",
)
print(result["memory"]["content_hash"])

Semantic search

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

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

# Usage — retrieve only memories from the researcher agent
results = await search_memory(
    query="API rate limits",
    tags=["agent:researcher"],
)
for mem in results:
    print(mem["content"], mem["tags"])

Store with deduplication bypass (conversation_id)

async def store_incremental(content: str, conversation_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/api/memories",
            json={
                "content": content,
                "tags": ["conversation"],
                "conversation_id": conversation_id,
            },
        )
        response.raise_for_status()
        return response.json()

List memories by tag

async def list_by_tag(tag: str, page: int = 1) -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/api/memories",
            params={"tags": tag, "page": page, "page_size": 20},
        )
        response.raise_for_status()
        return response.json()["memories"]

Query knowledge graph

async def get_associations(content_hash: str) -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/api/graph/associations/{content_hash}",
        )
        response.raise_for_status()
        return response.json()["associations"]

cURL Examples

# Store
curl -X POST http://localhost:8000/api/memories \
  -H "Content-Type: application/json" \
  -H "X-Agent-ID: researcher" \
  -d '{"content": "Deadline is March 15", "tags": ["project", "deadline"]}'

# Search
curl -X POST http://localhost:8000/api/memories/search \
  -H "Content-Type: application/json" \
  -d '{"query": "project deadlines", "limit": 5}'

# Search within agent scope
curl -X POST http://localhost:8000/api/memories/search \
  -H "Content-Type: application/json" \
  -d '{"query": "API limits", "tags": ["agent:researcher"]}'

# Health check
curl http://localhost:8000/api/health

SSE (Server-Sent Events) — Real-time Updates

Subscribe to memory events for reactive agent coordination:

import httpx

async def subscribe_to_memory_events():
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", f"{BASE_URL}/sse/events") as response:
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    import json
                    event_data = json.loads(line[5:].strip())
                    event_type = event_data.get("event_type")

                    if event_type == "memory_stored":
                        print(f"New memory: {event_data['content_hash']}")
                    elif event_type == "memory_deleted":
                        print(f"Deleted: {event_data['content_hash']}")

# Run in background task
import asyncio
asyncio.create_task(subscribe_to_memory_events())

X-Agent-ID Header

Any store request can include X-Agent-ID: <identifier> to automatically tag the memory:

# These two calls produce identical results:

# Explicit tag
await store_memory(content="...", tags=["agent:researcher", "api"])

# Header auto-tagging
await store_memory(content="...", tags=["api"], agent_id="researcher")
# Server appends "agent:researcher" automatically