use FilePath for documents

src/basic_memory/api/routers/documents.py

@@ -5,8 +5,7 @@ from typing import List
 from fastapi import APIRouter, HTTPException
 
 from basic_memory.deps import DocumentServiceDep
-from basic_memory.schemas.base import PathId
-from basic_memory.schemas.request import DocumentRequest
+from basic_memory.schemas.request import DocumentRequest, FilePath
 from basic_memory.schemas.response import DocumentResponse, DocumentCreateResponse
 from basic_memory.services.document_service import (
     DocumentNotFoundError,
@@ -33,7 +32,7 @@ async def create_document(
     """
     try:
         document = await service.create_document(
-            path=doc.path,
+            doc_path=doc.path,
             content=doc.content,
             metadata=doc.doc_metadata,
         )
@@ -51,39 +50,39 @@ async def list_documents(
     return [DocumentCreateResponse.model_validate(doc.__dict__) for doc in documents]
 
 
-@router.get("/{path_id:path}", response_model=DocumentResponse)
+@router.get("/{doc_path:path}", response_model=DocumentResponse)
 async def get_document(
-    path_id: PathId,
+    doc_path: FilePath,
     service: DocumentServiceDep,
 ) -> DocumentResponse:
     """Get a document by ID."""
     try:
-        document, content = await service.read_document_by_path(path_id)
+        document, content = await service.read_document_by_path(doc_path)
         doc_dict = document.__dict__ | {"content": content}
         response = DocumentResponse.model_validate(doc_dict)
         return response
     except DocumentNotFoundError:
-        raise HTTPException(status_code=404, detail=f"Document not found: {path_id}")
+        raise HTTPException(status_code=404, detail=f"Document not found: {doc_path}")
     except DocumentWriteError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
 
-@router.put("/{path_id:path}", response_model=DocumentResponse)
+@router.put("/{doc_path:path}", response_model=DocumentResponse)
 async def update_document(
-    path_id: PathId,
+    doc_path: FilePath,
     doc: DocumentRequest,
     service: DocumentServiceDep,
 ) -> DocumentResponse:
     """Update a document by ID."""
-    # Verify PathIds match
-    if doc.path != path_id:
+    # Verify FilePaths match
+    if doc.path != doc_path:
         raise HTTPException(
             status_code=400, detail="Document path in URL must match path in request body"
         )
 
     try:
         document = await service.update_document_by_path(
-            path_id=path_id,
+            path_id=doc_path,
             content=doc.content,
             metadata=doc.doc_metadata,
         )
@@ -95,14 +94,14 @@ async def update_document(
         raise HTTPException(status_code=400, detail=str(e))
 
 
-@router.delete("/{path_id:path}", status_code=204)
+@router.delete("/{doc_path:path}", status_code=204)
 async def delete_document(
-    path_id: PathId,
+    doc_path: FilePath,
     service: DocumentServiceDep,
 ) -> None:
     """Delete a document by ID."""
     try:
-        await service.delete_document_by_path(path_id)
+        await service.delete_document_by_path(doc_path)
     except DocumentNotFoundError:
         raise HTTPException(status_code=404, detail=f"Document not found: {id}")
     except DocumentWriteError as e:

src/basic_memory/mcp/server.py

@@ -15,6 +15,7 @@ mcp = FastMCP("Basic Memory")
 # Create shared async client
 client = AsyncClient(transport=ASGITransport(app=fastapi_app), base_url="http://test")
 
+
 def setup_logging(log_file: str = "basic-memory-mcp.log"):
     """Configure logging for the application."""
     # Remove default handler
@@ -41,6 +42,7 @@ def setup_logging(log_file: str = "basic-memory-mcp.log"):
         colorize=True,
     )
 
+
 async def log_api_call(method: str, url: str, data: Any, response: Any):
     """Log API request and response details."""
     logger.debug(f"API Request: {method} {url}")
@@ -48,10 +50,12 @@ async def log_api_call(method: str, url: str, data: Any, response: Any):
     logger.debug(f"Response Status: {response.status_code}")
     logger.debug(f"Response Data: {response.json()}")
 
+
 # Knowledge Graph Tools
 
 ## Create endpoints
 
+
 @mcp.tool()
 async def create_entities(entities: list[dict]) -> dict:
     """Create new entities in the knowledge graph."""
@@ -59,6 +63,7 @@ async def create_entities(entities: list[dict]) -> dict:
     await log_api_call("POST", "/knowledge/entities", entities, response)
     return response.json()
 
+
 @mcp.tool()
 async def create_relations(relations: list[dict]) -> dict:
     """Create relations between entities."""
@@ -66,6 +71,7 @@ async def create_relations(relations: list[dict]) -> dict:
     await log_api_call("POST", "/knowledge/relations", relations, response)
     return response.json()
 
+
 @mcp.tool()
 async def add_observations(path_id: str, observations: list[str]) -> dict:
     """Add observations to an entity."""
@@ -74,8 +80,10 @@ async def add_observations(path_id: str, observations: list[str]) -> dict:
     await log_api_call("POST", "/knowledge/observations", data, response)
     return response.json()
 
+
 ## Read endpoints
 
+
 @mcp.tool()
 async def get_entity(path_id: str) -> dict:
     """Get a specific entity by path_id."""
@@ -91,6 +99,7 @@ async def search_nodes(query: str) -> dict:
     await log_api_call("POST", "/knowledge/search", {"query": query}, response)
     return response.json()
 
+
 @mcp.tool()
 async def open_nodes(path_ids: List[str]) -> dict:
     """Search for entities in the knowledge graph."""
@@ -98,8 +107,10 @@ async def open_nodes(path_ids: List[str]) -> dict:
     await log_api_call("POST", "/knowledge/nodes", {"path_ids": path_ids}, response)
     return response.json()
 
+
 ## Delete endpoints
 
+
 @mcp.tool()
 async def delete_entities(path_ids: List[str]) -> dict:
     """Search for entities in the knowledge graph."""
@@ -107,18 +118,27 @@ async def delete_entities(path_ids: List[str]) -> dict:
     await log_api_call("POST", "/knowledge/entities/delete", {"path_ids": path_ids}, response)
     return response.json()
 
+
 @mcp.tool()
 async def delete_observations(path_id: str, observations: list[str]) -> dict:
     """Delete observations from an entity."""
-    data = {"path_id": path_id, "observations": observations}  # Match the parameter name with what we're using
-    response = await client.post("/knowledge/observations/delete", json=data)  # Change to observations endpoint
+    data = {
+        "path_id": path_id,
+        "observations": observations,
+    }  # Match the parameter name with what we're using
+    response = await client.post(
+        "/knowledge/observations/delete", json=data
+    )  # Change to observations endpoint
     await log_api_call("POST", "/knowledge/observations/delete", data, response)
     return response.json()
 
+
 @mcp.tool()
 async def delete_relations(relations: list[dict]) -> dict:
     """Delete relations between entities."""
-    response = await client.post("/knowledge/relations/delete", json={"relations": relations})  # Change to relations endpoint
+    response = await client.post(
+        "/knowledge/relations/delete", json={"relations": relations}
+    )  # Change to relations endpoint
     await log_api_call("POST", "/knowledge/relations/delete", {"relations": relations}, response)
     return response.json()
 
@@ -130,40 +150,46 @@ async def delete_relations(relations: list[dict]) -> dict:
 async def create_document(path: str, content: str, metadata: dict = None) -> dict:
     """Create a new document."""
     data = {"path": path, "content": content, "metadata": metadata}
-    response = await client.post("/documents", json=data)
-    await log_api_call("POST", "/documents", data, response)
+    response = await client.post("/documents/", json=data)
+    await log_api_call("POST", "/documents/", data, response)
     return response.json()
 
+
 @mcp.tool()
-async def get_document(path_id: str) -> dict:
+async def get_document(path: str) -> dict:
     """Get a document by path_id."""
-    response = await client.get(f"/documents/{path_id}")
-    await log_api_call("GET", f"/documents/{path_id}", None, response)
+    response = await client.get(f"/documents/{path}/")
+    await log_api_call("GET", f"/documents/{path}/", None, response)
     return response.json()
 
+
 @mcp.tool()
-async def update_document(path_id: str, content: str, metadata: dict = None) -> dict:
+async def update_document(path: str, content: str, metadata: dict = None) -> dict:
     """Update an existing document."""
-    data = {"path": path_id, "content": content, "metadata": metadata}
-    response = await client.put(f"/documents/{path_id}", json=data)
-    await log_api_call("PUT", f"/documents/{path_id}", data, response)
+    data = {"path": path, "content": content, "metadata": metadata}
+    response = await client.put(f"/documents/{path}/", json=data)
+    await log_api_call("PUT", f"/documents/{path}/", data, response)
     return response.json()
 
+
 @mcp.tool()
 async def list_documents() -> list:
     """List all documents."""
-    response = await client.get("/documents")
-    await log_api_call("GET", "/documents", None, response)
+    response = await client.get("/documents/")
+    await log_api_call("GET", "/documents/", None, response)
     return response.json()
 
+
 @mcp.tool()
-async def delete_document(path_id: str) -> dict:
-    """Update an existing document."""
-    response = await client.put(f"/documents/{path_id}")
-    await log_api_call("DELETE", f"/documents/{path_id}", None, response)
-    return response.json()
+async def delete_document(path: str) -> dict:
+    """Delete an existing document."""
+    response = await client.delete(f"/documents/{path}/")
+    await log_api_call("DELETE", f"/documents/{path}/", None, response)
+    if response.status_code == 204:
+        return {"deleted": True}
+
 
 if __name__ == "__main__":
     setup_logging()
     logger.info("Starting Basic Memory MCP server")
-    mcp.run()
+    mcp.run()

src/basic_memory/schemas/base.py

@@ -61,7 +61,7 @@ def to_snake_case(name: str) -> str:
 
 
 def validate_path_format(path: str) -> str:
-    """Validate path has the correct format: type/name."""
+    """Validate path has the correct format: not empty."""
     if not path or not isinstance(path, str):
         raise ValueError("Path must be a non-empty string")
 

src/basic_memory/schemas/request.py

@@ -1,9 +1,10 @@
 """Request schemas for interacting with the knowledge graph."""
 
 from typing import List, Optional, Annotated, Dict, Any
+from annotated_types import MaxLen, MinLen
+from pydantic.json_schema import Pattern
 
-from annotated_types import MinLen, MaxLen
-from pydantic import BaseModel
+from pydantic import BaseModel, StringConstraints
 
 from basic_memory.schemas.base import Observation, Entity, Relation, PathId
 
@@ -201,7 +202,15 @@ class CreateRelationsRequest(BaseModel):
 ## document
 
 
+FilePath = Annotated[
+    str,
+    StringConstraints(pattern=r'^[a-zA-Z0-9_/.-]+\.md$'),
+    MinLen(1),
+    MaxLen(255)
+]
+
+
 class DocumentRequest(BaseModel):
-    path: PathId
+    path: FilePath
     content: str
     doc_metadata: Optional[Dict[str, Any]] = None

src/basic_memory/services/document_service.py

@@ -96,13 +96,13 @@ class DocumentService(BaseService[DocumentRepository]):
         return await self.repository.find_all()
 
     async def create_document(
-        self, path: str, content: str, metadata: Optional[Dict[str, Any]] = None
+        self, doc_path: str, content: str, metadata: Optional[Dict[str, Any]] = None
     ) -> Document:
         """
         Create a new document.
 
         Args:
-            path: Path where to create the document
+            doc_path: Path where to create the document
             content: Document content
             metadata: Optional metadata to store
 
@@ -112,20 +112,20 @@ class DocumentService(BaseService[DocumentRepository]):
         Raises:
             DocumentWriteError: If file cannot be written
         """
-        logger.debug(f"Creating document at path: {path}")
+        logger.debug(f"Creating document at path: {doc_path}")
 
         # Ensure parent directories exist
-        file_path = self.get_document_path(path)
+        file_path = self.get_document_path(doc_path)
         await self.ensure_parent_directory(file_path)
 
         # db reference
         document = None
         try:
-            # 1. Create initial DB record to get ID
-            document = await self.repository.create({"path": str(path), "doc_metadata": metadata})
+            # 1. Create initial DB record to get row id
+            document = await self.repository.create({"path": str(doc_path), "doc_metadata": metadata})
 
-            # 2. Add frontmatter with DB-generated ID
-            content_with_frontmatter = await self.add_frontmatter(content, path, metadata)
+            # 2. Add frontmatter with path_id
+            content_with_frontmatter = await self.add_frontmatter(content, doc_path, metadata)
 
             # 3. Write complete file
             file_path.write_text(content_with_frontmatter)