mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
delete old mcp tests, add docs to tools
This commit is contained in:
phernandez
+195
-28
@@ -1,7 +1,7 @@
|
||||
"""Basic Memory MCP server using fastmcp - proxies to FastAPI endpoints."""
|
||||
|
||||
import sys
|
||||
from typing import Any, List
|
||||
from typing import Any, List, Optional, Dict
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from loguru import logger
|
||||
@@ -54,8 +54,24 @@ async def log_api_call(method: str, url: str, data: Any, response: Any):
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def create_entities(entities: list[dict]) -> dict:
|
||||
"""Create new entities in the knowledge graph."""
|
||||
async def create_entities(entities: List[dict]) -> dict:
|
||||
"""Create new entities in the knowledge graph.
|
||||
|
||||
Args:
|
||||
entities: List of entity dictionaries, each containing:
|
||||
- name: Entity name
|
||||
- entity_type: Classification (e.g., 'component', 'specification')
|
||||
- description: Optional description
|
||||
- observations: Optional list of initial observations
|
||||
|
||||
Example:
|
||||
create_entities([{
|
||||
"name": "Knowledge Format",
|
||||
"entity_type": "specification",
|
||||
"description": "Document format specification",
|
||||
"observations": ["Uses markdown format", "Supports YAML frontmatter"]
|
||||
}])
|
||||
"""
|
||||
url = "/knowledge/entities"
|
||||
data = {"entities": entities}
|
||||
response = await client.post(url, json=data)
|
||||
@@ -63,8 +79,24 @@ async def create_entities(entities: list[dict]) -> dict:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def create_relations(relations: list[dict]) -> dict:
|
||||
"""Create relations between entities."""
|
||||
async def create_relations(relations: List[dict]) -> dict:
|
||||
"""Create relations between existing entities.
|
||||
|
||||
Args:
|
||||
relations: List of relation dictionaries, each containing:
|
||||
- from_id: Source entity path_id
|
||||
- to_id: Target entity path_id
|
||||
- relation_type: Type of relationship in active voice
|
||||
- context: Optional context for the relation
|
||||
|
||||
Example:
|
||||
create_relations([{
|
||||
"from_id": "test/parser_test",
|
||||
"to_id": "component/parser",
|
||||
"relation_type": "validates",
|
||||
"context": "Unit test coverage"
|
||||
}])
|
||||
"""
|
||||
url = "/knowledge/relations"
|
||||
data = {"relations": relations}
|
||||
response = await client.post(url, json=data)
|
||||
@@ -72,10 +104,32 @@ async def create_relations(relations: list[dict]) -> dict:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def add_observations(path_id: str, observations: list[str]) -> dict:
|
||||
"""Add observations to an entity."""
|
||||
async def add_observations(path_id: str, observations: List[dict], context: Optional[str] = None) -> dict:
|
||||
"""Add observations to an entity.
|
||||
|
||||
Args:
|
||||
path_id: Entity path ID
|
||||
observations: List of observations, each containing:
|
||||
- content: The observation text
|
||||
- category: Optional category ('tech', 'design', 'feature', 'note', 'issue', 'todo')
|
||||
context: Optional shared context for all observations
|
||||
|
||||
Example:
|
||||
add_observations(
|
||||
"specification/knowledge_format",
|
||||
observations=[
|
||||
{"content": "Uses markdown format", "category": "tech"},
|
||||
{"content": "Designed for readability", "category": "design"}
|
||||
],
|
||||
context="Initial design"
|
||||
)
|
||||
"""
|
||||
url = "/knowledge/observations"
|
||||
data = {"path_id": path_id, "observations": observations}
|
||||
data = {
|
||||
"path_id": path_id,
|
||||
"observations": observations,
|
||||
"context": context
|
||||
}
|
||||
response = await client.post(url, json=data)
|
||||
await log_api_call("POST", url, data, response)
|
||||
return response.json()
|
||||
@@ -86,24 +140,59 @@ async def add_observations(path_id: str, observations: list[str]) -> dict:
|
||||
|
||||
@mcp.tool()
|
||||
async def get_entity(path_id: str) -> dict:
|
||||
"""Get a specific entity by path_id."""
|
||||
"""Get a specific entity by its path_id.
|
||||
|
||||
Args:
|
||||
path_id: Entity path ID (e.g., 'component/memory_service')
|
||||
|
||||
Returns:
|
||||
Complete entity information including observations and relations.
|
||||
|
||||
Example:
|
||||
get_entity("specification/knowledge_format")
|
||||
"""
|
||||
url = f"/knowledge/entities/{path_id}"
|
||||
response = await client.get(url)
|
||||
return response.json()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def search_nodes(query: str) -> dict:
|
||||
"""Search for entities in the knowledge graph."""
|
||||
async def search_nodes(query: str, category: Optional[str] = None) -> dict:
|
||||
"""Search for entities in the knowledge graph.
|
||||
|
||||
Args:
|
||||
query: Search text to match against entities
|
||||
category: Optional category to filter observations by
|
||||
|
||||
Returns:
|
||||
Matching entities with their observations and relations.
|
||||
|
||||
Example:
|
||||
search_nodes("markdown format", category="tech") # Find tech observations about markdown
|
||||
search_nodes("implementation") # Search all categories
|
||||
"""
|
||||
url = "/knowledge/search"
|
||||
data = {"query": query}
|
||||
data = {"query": query, "category": category}
|
||||
response = await client.post(url, json=data)
|
||||
return response.json()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def open_nodes(path_ids: List[str]) -> dict:
|
||||
"""Search for entities in the knowledge graph."""
|
||||
"""Load multiple entities by their path_ids.
|
||||
|
||||
Args:
|
||||
path_ids: List of entity path IDs to load
|
||||
|
||||
Returns:
|
||||
Dictionary of loaded entities with their observations and relations.
|
||||
|
||||
Example:
|
||||
open_nodes([
|
||||
"specification/knowledge_format",
|
||||
"component/parser"
|
||||
])
|
||||
"""
|
||||
url = "/knowledge/nodes"
|
||||
data = {"path_ids": path_ids}
|
||||
response = await client.post(url, json=data)
|
||||
@@ -115,7 +204,14 @@ async def open_nodes(path_ids: List[str]) -> dict:
|
||||
|
||||
@mcp.tool()
|
||||
async def delete_entities(path_ids: List[str]) -> dict:
|
||||
"""Search for entities in the knowledge graph."""
|
||||
"""Delete entities from the knowledge graph.
|
||||
|
||||
Args:
|
||||
path_ids: List of entity path IDs to delete
|
||||
|
||||
Example:
|
||||
delete_entities(["test/obsolete_test", "component/old_component"])
|
||||
"""
|
||||
url = "/knowledge/entities/delete"
|
||||
data = {"path_ids": path_ids}
|
||||
response = await client.post(url, json=data)
|
||||
@@ -123,8 +219,19 @@ async def delete_entities(path_ids: List[str]) -> dict:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def delete_observations(path_id: str, observations: list[str]) -> dict:
|
||||
"""Delete observations from an entity."""
|
||||
async def delete_observations(path_id: str, observations: List[str]) -> dict:
|
||||
"""Delete specific observations from an entity.
|
||||
|
||||
Args:
|
||||
path_id: Entity path ID
|
||||
observations: List of observation content strings to delete
|
||||
|
||||
Example:
|
||||
delete_observations(
|
||||
"component/parser",
|
||||
["Obsolete implementation detail", "Old design note"]
|
||||
)
|
||||
"""
|
||||
url = "/knowledge/observations/delete"
|
||||
data = {
|
||||
"path_id": path_id,
|
||||
@@ -132,18 +239,32 @@ async def delete_observations(path_id: str, observations: list[str]) -> dict:
|
||||
}
|
||||
response = await client.post(
|
||||
url, json=data
|
||||
)
|
||||
)
|
||||
return response.json()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def delete_relations(relations: list[dict]) -> dict:
|
||||
"""Delete relations between entities."""
|
||||
async def delete_relations(relations: List[dict]) -> dict:
|
||||
"""Delete relations between entities.
|
||||
|
||||
Args:
|
||||
relations: List of relation dictionaries to delete, each containing:
|
||||
- from_id: Source entity path_id
|
||||
- to_id: Target entity path_id
|
||||
- relation_type: Type of relationship
|
||||
|
||||
Example:
|
||||
delete_relations([{
|
||||
"from_id": "test/old_test",
|
||||
"to_id": "component/parser",
|
||||
"relation_type": "validates"
|
||||
}])
|
||||
"""
|
||||
url = "/knowledge/relations/delete"
|
||||
data = {"relations": relations}
|
||||
response = await client.post(
|
||||
url, json=data
|
||||
)
|
||||
)
|
||||
return response.json()
|
||||
|
||||
|
||||
@@ -151,8 +272,21 @@ async def delete_relations(relations: list[dict]) -> dict:
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def create_document(path: str, content: str, metadata: dict = None) -> dict:
|
||||
"""Create a new document."""
|
||||
async def create_document(path: str, content: str, metadata: Optional[Dict] = None) -> dict:
|
||||
"""Create a new document.
|
||||
|
||||
Args:
|
||||
path: Document path (must end in .md)
|
||||
content: Document content as markdown text
|
||||
metadata: Optional metadata dictionary
|
||||
|
||||
Example:
|
||||
create_document(
|
||||
"docs/format.md",
|
||||
"# Format Specification\n\nDetails here...",
|
||||
metadata={"author": "AI team"}
|
||||
)
|
||||
"""
|
||||
url = "/documents/create"
|
||||
data = {"path": path, "content": content, "metadata": metadata}
|
||||
response = await client.post(url, json=data)
|
||||
@@ -161,15 +295,35 @@ async def create_document(path: str, content: str, metadata: dict = None) -> dic
|
||||
|
||||
@mcp.tool()
|
||||
async def get_document(path: str) -> dict:
|
||||
"""Get a document by path_id."""
|
||||
"""Get a document by its path.
|
||||
|
||||
Args:
|
||||
path: Document path to retrieve
|
||||
|
||||
Example:
|
||||
get_document("docs/format.md")
|
||||
"""
|
||||
url = f"/documents/{path}"
|
||||
response = await client.get(url)
|
||||
return response.json()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def update_document(path: str, content: str, metadata: dict = None) -> dict:
|
||||
"""Update an existing document."""
|
||||
async def update_document(path: str, content: str, metadata: Optional[Dict] = None) -> dict:
|
||||
"""Update an existing document.
|
||||
|
||||
Args:
|
||||
path: Document path
|
||||
content: New document content
|
||||
metadata: Optional new metadata
|
||||
|
||||
Example:
|
||||
update_document(
|
||||
"docs/format.md",
|
||||
"# Updated Format\n\nNew content...",
|
||||
metadata={"updated_by": "AI team"}
|
||||
)
|
||||
"""
|
||||
url = f"/documents/{path}"
|
||||
data = {"path": path, "content": content, "metadata": metadata}
|
||||
response = await client.put(url, json=data)
|
||||
@@ -178,7 +332,14 @@ async def update_document(path: str, content: str, metadata: dict = None) -> dic
|
||||
|
||||
@mcp.tool()
|
||||
async def list_documents() -> list:
|
||||
"""List all documents."""
|
||||
"""List all documents in the system.
|
||||
|
||||
Returns:
|
||||
List of document paths and metadata.
|
||||
|
||||
Example:
|
||||
list_documents() # Get all document paths
|
||||
"""
|
||||
url = "/documents/list"
|
||||
response = await client.get(url)
|
||||
return response.json()
|
||||
@@ -186,13 +347,19 @@ async def list_documents() -> list:
|
||||
|
||||
@mcp.tool()
|
||||
async def delete_document(path: str) -> dict:
|
||||
"""Delete an existing document."""
|
||||
"""Delete a document.
|
||||
|
||||
Args:
|
||||
path: Path of document to delete
|
||||
|
||||
Example:
|
||||
delete_document("docs/obsolete.md")
|
||||
"""
|
||||
url = f"/documents/{path}"
|
||||
response = await client.delete(url)
|
||||
if response.status_code == 204:
|
||||
return {"deleted": True}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
setup_logging()
|
||||
logger.info("Starting Basic Memory MCP server")
|
||||
|
||||
@@ -1,40 +0,0 @@
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, EntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_add_observations(app, test_entity_data, client):
|
||||
# """Test adding observations to an existing entity."""
|
||||
#
|
||||
# # First create an entity
|
||||
# create_result = await handle_call_tool("create_entities", test_entity_data)
|
||||
# create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# entity_id = create_response.entities[0].id
|
||||
#
|
||||
# # Add new observation
|
||||
# result = await handle_call_tool(
|
||||
# "add_observations", {"entity_id": entity_id, "observations": ["A new observation"]}
|
||||
# )
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
#
|
||||
# # Verify observation was added
|
||||
# response = EntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert response.id == entity_id
|
||||
# assert len(response.observations) == 2 # 1 already present
|
||||
# assert response.observations[1].content == "A new observation"
|
||||
#
|
||||
# # Verify through API
|
||||
# api_response = await client.get(f"/knowledge/entities/{entity_id}")
|
||||
# assert api_response.status_code == 200
|
||||
# entity = api_response.json()
|
||||
# assert len(entity["observations"]) == 2 # Original + new
|
||||
# assert "A new observation" in [o["content"] for o in entity["observations"]]
|
||||
@@ -1,107 +0,0 @@
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_single_entity(app):
|
||||
# """Test creating a single entity."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "SingleTest", "entity_type": "test", "observations": ["Test observation"]}
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify entity creation
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.entities) == 1
|
||||
# entity = response.entities[0]
|
||||
# assert entity.name == "SingleTest"
|
||||
# assert entity.entity_type == "test"
|
||||
# assert len(entity.observations) == 1
|
||||
# assert entity.observations[0].content == "Test observation"
|
||||
# assert entity.id is not None
|
||||
#
|
||||
# # Verify entity can be found via search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "SingleTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert search_response.matches[0].name == "SingleTest"
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_multiple_entities(app):
|
||||
# """Test creating multiple entities in one call."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "BulkTest1", "entity_type": "test", "observations": ["First bulk test"]},
|
||||
# {"name": "BulkTest2", "entity_type": "test", "observations": ["Second bulk test"]},
|
||||
# {"name": "BulkTest3", "entity_type": "demo", "observations": ["Third bulk test"]},
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
#
|
||||
# # Verify response
|
||||
# assert len(result) == 1
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Verify all entities were created
|
||||
# assert len(response.entities) == 3
|
||||
#
|
||||
# # Check specific entities
|
||||
# entities = {e.name: e for e in response.entities}
|
||||
# assert "BulkTest1" in entities
|
||||
# assert "BulkTest2" in entities
|
||||
# assert "BulkTest3" in entities
|
||||
#
|
||||
# # Verify IDs were generated correctly
|
||||
# assert entities["BulkTest1"].id is not None
|
||||
# assert entities["BulkTest2"].id is not None
|
||||
# assert entities["BulkTest3"].id is not None
|
||||
#
|
||||
# # Verify observations were saved
|
||||
# assert len(entities["BulkTest1"].observations) == 1
|
||||
# assert entities["BulkTest1"].observations[0].content == "First bulk test"
|
||||
#
|
||||
# # Verify entities can be found via search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "BulkTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(search_response.matches) == 3
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_entity_with_all_fields(app):
|
||||
# """Test creating entity with all possible fields populated."""
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "FullEntity",
|
||||
# "entity_type": "test",
|
||||
# "description": "A complete test entity",
|
||||
# "observations": ["First observation", "Second observation"],
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_entities", entity_data)
|
||||
# response = CreateEntityResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# entity = response.entities[0]
|
||||
# assert entity.name == "FullEntity"
|
||||
# assert entity.description == "A complete test entity"
|
||||
# assert len(entity.observations) == 2
|
||||
# assert entity.observations[0].content == "First observation"
|
||||
# assert entity.observations[1].content == "Second observation"
|
||||
@@ -1,48 +0,0 @@
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_create_relations(app):
|
||||
# """Test creating relations between entities."""
|
||||
# # Create two test entities
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {"name": "TestEntityA", "entity_type": "test", "observations": ["Entity A"]},
|
||||
# {"name": "TestEntityB", "entity_type": "test", "observations": ["Entity B"]},
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# from_entity = create_entity_response.entities[0]
|
||||
# to_entity = create_entity_response.entities[1]
|
||||
# # Create relation between them
|
||||
# relation_data = {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# result = await handle_call_tool("create_relations", relation_data)
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "TestEntityA"})
|
||||
# response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# assert len(response.matches) == 1
|
||||
# entity = response.matches[0]
|
||||
# assert len(entity.relations) == 1
|
||||
# assert entity.relations[0].to_id == to_entity.id
|
||||
# assert entity.relations[0].relation_type == "relates_to"
|
||||
@@ -1,35 +0,0 @@
|
||||
# """Tests for MCP delete_entities tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_entities(app):
|
||||
# """Test deleting entities."""
|
||||
# # Create test entities
|
||||
# entities = {
|
||||
# "entities": [
|
||||
# {"name": "DeleteTest1", "entity_type": "test", "observations": ["To be deleted 1"]},
|
||||
# {"name": "DeleteTest2", "entity_type": "test", "observations": ["To be deleted 2"]},
|
||||
# ]
|
||||
# }
|
||||
# create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# # Delete first entity
|
||||
# await handle_call_tool(
|
||||
# "delete_entities", {"entity_ids": [create_entity_response.entities[0].id]}
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "DeleteTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Only second entity should remain
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert search_response.matches[0].name == "DeleteTest2"
|
||||
@@ -1,47 +0,0 @@
|
||||
# """Tests for MCP delete_observations tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import CreateEntityResponse, SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_observations(app):
|
||||
# """Test deleting specific observations from an entity."""
|
||||
# # Create entity with multiple observations
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "ObsDeleteTest",
|
||||
# "entity_type": "test",
|
||||
# "observations": [
|
||||
# "Keep this observation",
|
||||
# "Delete this observation",
|
||||
# "Also keep this",
|
||||
# ],
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
# create_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_response = CreateEntityResponse.model_validate_json(create_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# entity_id = create_response.entities[0].id
|
||||
#
|
||||
# # Delete specific observation
|
||||
# await handle_call_tool(
|
||||
# "delete_observations", {"entity_id": entity_id, "deletions": ["Delete this observation"]}
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "ObsDeleteTest"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(
|
||||
# search_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
#
|
||||
# # Check remaining observations
|
||||
# entity = search_response.matches[0]
|
||||
# observations = [o.content for o in entity.observations]
|
||||
# assert len(observations) == 2
|
||||
# assert "Delete this observation" not in observations
|
||||
# assert "Keep this observation" in observations
|
||||
# assert "Also keep this" in observations
|
||||
@@ -1,58 +0,0 @@
|
||||
# """Tests for MCP delete_relations tool."""
|
||||
#
|
||||
# import pytest
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_delete_relations(app):
|
||||
# """Test deleting relations between entities."""
|
||||
# # Create test entities with relation
|
||||
# entities = {
|
||||
# "entities": [
|
||||
# {"name": "RelSource", "entity_type": "test", "observations": ["Source entity"]},
|
||||
# {"name": "RelTarget", "entity_type": "test", "observations": ["Target entity"]},
|
||||
# ]
|
||||
# }
|
||||
# create_entity_result = await handle_call_tool("create_entities", entities)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
# from_entity = create_entity_response.entities[0]
|
||||
# to_entity = create_entity_response.entities[1]
|
||||
#
|
||||
# # Create relation
|
||||
# relation = {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# }
|
||||
# await handle_call_tool("create_relations", relation)
|
||||
#
|
||||
# # Delete the relation
|
||||
# await handle_call_tool(
|
||||
# "delete_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": from_entity.id,
|
||||
# "to_id": to_entity.id,
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
#
|
||||
# # Verify through search
|
||||
# search_result = await handle_call_tool("search_nodes", {"query": "relsource"})
|
||||
# search_response = SearchNodesResponse.model_validate_json(search_result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
#
|
||||
# # Source entity should exist but have no relations
|
||||
# assert len(search_response.matches) == 1
|
||||
# assert len(search_response.matches[0].relations) == 0
|
||||
@@ -1,242 +0,0 @@
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.shared.exceptions import McpError
|
||||
# from mcp.types import INVALID_PARAMS, METHOD_NOT_FOUND
|
||||
#
|
||||
# from basic_memory.mcp.server import handle_call_tool
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_tool_name(app):
|
||||
# """Test calling a non-existent tool."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("not_a_tool", {})
|
||||
# assert "Unknown tool" in str(exc.value)
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_missing_required_field(app):
|
||||
# """Test validation when required fields are missing."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("search_nodes", {})
|
||||
# assert "query" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {})
|
||||
# assert "entities" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_document", {})
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
# error_msg = str(exc.value).lower()
|
||||
# assert "path" in error_msg or "content" in error_msg
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_empty_arrays(app):
|
||||
# """Test validation of array fields that can't be empty."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {"entities": []})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("open_nodes", {"entity_ids": []})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_field_types(app):
|
||||
# """Test validation when fields have wrong types."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("search_nodes", {"query": 123})
|
||||
# assert "string" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("create_entities", {"entities": "not an array"})
|
||||
# assert "array" in str(exc.value).lower() or "list" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("get_document", {"id": "not an integer"})
|
||||
# assert "integer" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_nested_fields(app):
|
||||
# """Test validation of nested object fields."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_entities",
|
||||
# {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "Test",
|
||||
# # Missing required entity_type
|
||||
# "observations": [],
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "entity_type" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "test.md",
|
||||
# "content": "test",
|
||||
# "doc_metadata": "not an object" # Should be dict/null
|
||||
# },
|
||||
# )
|
||||
# assert "doc_metadata" in str(exc.value).lower()
|
||||
# assert exc.value.args[0] == INVALID_PARAMS
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_relation_format_to_id(app):
|
||||
# """Test validation of relation data."""
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": 1,
|
||||
# # Missing to_id
|
||||
# "relation_type": "relates_to",
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "to_id" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_invalid_relation_format_relation_type(app):
|
||||
# # Invalid relation type
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_relations",
|
||||
# {
|
||||
# "relations": [
|
||||
# {
|
||||
# "from_id": 1,
|
||||
# "to_id": 2,
|
||||
# "relation_type": "", # Empty relation type
|
||||
# }
|
||||
# ]
|
||||
# },
|
||||
# )
|
||||
# assert "relation_type" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_observation_validation_len(app):
|
||||
# """Test validation specific to observations."""
|
||||
# # Empty observations
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "add_observations",
|
||||
# {
|
||||
# "entity_id": 1,
|
||||
# "observations": ["", ""], # Empty observations
|
||||
# },
|
||||
# )
|
||||
# assert "observations" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_observation_validation_delete(app):
|
||||
# # Empty deletions
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "delete_observations",
|
||||
# {
|
||||
# "entity_id": 1,
|
||||
# "deletions": [], # Empty deletions
|
||||
# },
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_edge_case_validation_search_len(app):
|
||||
# """Test edge cases in validation."""
|
||||
# # Very long strings
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "search_nodes",
|
||||
# {"query": "x" * 10000}, # Extremely long query
|
||||
# )
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_document_endpoint_validation(app):
|
||||
# """Test validation specific to document endpoints."""
|
||||
# # Invalid ID format for get_document
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool("get_document", {"id": -1})
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "greater than 0" in str(exc.value).lower()
|
||||
#
|
||||
# # Invalid document path
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "", # Empty path
|
||||
# "content": "test content"
|
||||
# }
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "path" in str(exc.value).lower()
|
||||
#
|
||||
# # Update without content
|
||||
# with pytest.raises(McpError) as exc:
|
||||
# await handle_call_tool(
|
||||
# "update_document",
|
||||
# {
|
||||
# "id": 1,
|
||||
# "doc_metadata": None
|
||||
# }
|
||||
# )
|
||||
# assert INVALID_PARAMS == exc.value.args[0]
|
||||
# assert "content" in str(exc.value).lower()
|
||||
#
|
||||
#
|
||||
# # We'll skip this test for now since it requires database setup
|
||||
# @pytest.mark.skip(reason="Requires database setup")
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_document_http_methods(app):
|
||||
# """Test that document endpoints use correct HTTP methods."""
|
||||
# # Test GET endpoints
|
||||
# await handle_call_tool("list_documents", {})
|
||||
# await handle_call_tool("get_document", {"id": 1})
|
||||
#
|
||||
# # Test POST endpoint
|
||||
# response = await handle_call_tool(
|
||||
# "create_document",
|
||||
# {
|
||||
# "path": "test.md",
|
||||
# "content": "test content"
|
||||
# }
|
||||
# )
|
||||
#
|
||||
# # Test PUT endpoint
|
||||
# await handle_call_tool(
|
||||
# "update_document",
|
||||
# {
|
||||
# "id": 1,
|
||||
# "content": "updated content",
|
||||
# "doc_metadata": None
|
||||
# }
|
||||
# )
|
||||
#
|
||||
# # Test DELETE endpoint
|
||||
# await handle_call_tool("delete_document", {"id": 1})
|
||||
@@ -1,63 +0,0 @@
|
||||
# """Tests for MCP open_nodes tool."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import OpenNodesResponse, CreateEntityResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_open_nodes(app):
|
||||
# """Test retrieving specific nodes by name."""
|
||||
# # Create test entities
|
||||
# entity_data = {
|
||||
# "entities": [
|
||||
# {
|
||||
# "name": "OpenTestA",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["First test entity"],
|
||||
# },
|
||||
# {
|
||||
# "name": "OpenTestB",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["Second test entity"],
|
||||
# },
|
||||
# {
|
||||
# "name": "OpenTestC",
|
||||
# "entity_type": "test",
|
||||
# "observations": ["Third test entity"],
|
||||
# },
|
||||
# ]
|
||||
# }
|
||||
#
|
||||
# create_entity_result = await handle_call_tool("create_entities", entity_data)
|
||||
# create_entity_response = CreateEntityResponse.model_validate_json(
|
||||
# create_entity_result[0].resource.text # pyright: ignore [reportAttributeAccessIssue]
|
||||
# )
|
||||
# entity_a = create_entity_response.entities[0]
|
||||
# entity_b = create_entity_response.entities[1]
|
||||
#
|
||||
# # Open specific nodes
|
||||
# result = await handle_call_tool("open_nodes", {"entity_ids": [entity_a.id, entity_b.id]})
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify entities returned
|
||||
# response = OpenNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.entities) == 2
|
||||
#
|
||||
# # Entities should be returned in same order as requested
|
||||
# assert response.entities[0].name == "OpenTestA"
|
||||
# assert response.entities[1].name == "OpenTestB"
|
||||
#
|
||||
# # Verify entity content
|
||||
# entity = response.entities[0]
|
||||
# assert entity.id == entity_a.id
|
||||
# assert entity.entity_type == "test"
|
||||
# assert len(entity.observations) == 1
|
||||
# assert entity.observations[0].content == "First test entity"
|
||||
@@ -1,37 +0,0 @@
|
||||
# """Tests for the MCP server implementation using FastAPI TestClient."""
|
||||
#
|
||||
# import pytest
|
||||
# from mcp.types import EmbeddedResource
|
||||
#
|
||||
# from basic_memory.mcp.server import MIME_TYPE, handle_call_tool
|
||||
# from basic_memory.schemas import SearchNodesResponse
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_search_nodes(app, test_entity_data, client):
|
||||
# """Test searching for an entity after creating it."""
|
||||
#
|
||||
# # First create an entity
|
||||
# await handle_call_tool("create_entities", test_entity_data)
|
||||
#
|
||||
# # Then search for it
|
||||
# result = await handle_call_tool("search_nodes", {"query": "Test Entity"})
|
||||
#
|
||||
# # Verify response format
|
||||
# assert len(result) == 1
|
||||
# assert isinstance(result[0], EmbeddedResource)
|
||||
# assert result[0].type == "resource"
|
||||
# assert result[0].resource.mimeType == MIME_TYPE
|
||||
#
|
||||
# # Verify search results
|
||||
# response = SearchNodesResponse.model_validate_json(result[0].resource.text) # pyright: ignore [reportAttributeAccessIssue]
|
||||
# assert len(response.matches) == 1
|
||||
# assert response.matches[0].name == "Test Entity"
|
||||
# assert response.query == "Test Entity"
|
||||
#
|
||||
# # Verify through API
|
||||
# api_response = await client.post("/knowledge/search", json={"query": "Test Entity"})
|
||||
# assert api_response.status_code == 200
|
||||
# data = api_response.json()
|
||||
# assert len(data["matches"]) == 1
|
||||
# assert data["matches"][0]["name"] == "Test Entity"
|
||||
@@ -1,25 +0,0 @@
|
||||
# """Test to show MCP tool documentation."""
|
||||
#
|
||||
# import json
|
||||
# import pytest
|
||||
# from mcp.types import Tool
|
||||
# from basic_memory.mcp.server import handle_list_tools
|
||||
#
|
||||
#
|
||||
# @pytest.mark.asyncio
|
||||
# async def test_list_tools():
|
||||
# """List available tools and their documentation."""
|
||||
# tools = await handle_list_tools()
|
||||
# assert isinstance(tools, list)
|
||||
# assert all(isinstance(t, Tool) for t in tools)
|
||||
#
|
||||
# print("\nAvailable MCP Tools:\n")
|
||||
#
|
||||
# # Print each tool's documentation
|
||||
# for tool in tools:
|
||||
# print(f"Tool: {tool.name}")
|
||||
# print(f"Description: {tool.description}")
|
||||
# print("Required fields:", tool.inputSchema.get("required", []))
|
||||
# print()
|
||||
# print("Schema:", json.dumps(tool.inputSchema, indent=2))
|
||||
# print("-" * 80 + "\n")
|
||||
Reference in New Issue
Block a user