From fae82c4cc7a786f87c16986e979823334336acba Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 31 Dec 2024 00:19:33 -0600 Subject: [PATCH] update tool schema docs --- src/basic_memory/mcp/tools/activity.py | 185 ++++++++-- src/basic_memory/mcp/tools/discovery.py | 244 ++++++++++--- src/basic_memory/mcp/tools/documents.py | 444 +++++++++++++++++------- src/basic_memory/mcp/tools/help.py | 130 ++++++- src/basic_memory/mcp/tools/knowledge.py | 423 +++++++++++++--------- src/basic_memory/mcp/tools/search.py | 287 +++++++++++---- 6 files changed, 1280 insertions(+), 433 deletions(-) diff --git a/src/basic_memory/mcp/tools/activity.py b/src/basic_memory/mcp/tools/activity.py index 8a3c3d54..0dfedc65 100644 --- a/src/basic_memory/mcp/tools/activity.py +++ b/src/basic_memory/mcp/tools/activity.py @@ -9,39 +9,174 @@ from basic_memory.mcp.server import mcp from basic_memory.schemas.activity import ActivityType, RecentActivity -@mcp.tool() +@mcp.tool( + description=""" + Get recent activity across your knowledge base. + + This tool provides a comprehensive view of changes across your knowledge base, + including document modifications, entity updates, and relationship changes. + It supports flexible time ranges and filtering by activity type. + + The activity log helps you: + - Track recent changes to your knowledge base + - Monitor document and entity modifications + - Understand system usage patterns + - Identify most active areas + + Activity is tracked for: + - Document changes (creation, updates, deletion) + - Entity modifications + - Relation changes between entities + """, + examples=[ + { + "name": "Daily Changes Overview", + "description": "Get a summary of all changes in the last day", + "code": """ +# Get last 24 hours of activity +activity = await get_recent_activity() + +# Print summary statistics +print(f"Total changes: {len(activity.changes)}") +print(f"Documents modified: {activity.summary.document_changes}") +print(f"Entities modified: {activity.summary.entity_changes}") +print(f"Relations changed: {activity.summary.relation_changes}") + +# Show most active areas +print("\\nMost active paths:") +for path in activity.summary.most_active_paths: + print(f"- {path}") +""" + }, + { + "name": "Filter Document Changes", + "description": "Focus on recent document activity", + "code": """ +# Get only document changes from last hour +docs = await get_recent_activity( + timeframe="1h", + activity_types=[ActivityType.DOCUMENT] +) + +# Show document changes chronologically +for change in sorted(docs.changes, key=lambda x: x.timestamp): + print(f"{change.timestamp}: {change.path_id}") + print(f" {change.change_type}: {change.summary}") +""" + }, + { + "name": "Weekly Activity Analysis", + "description": "Analyze patterns over past week", + "code": """ +# Get full week of activity +weekly = await get_recent_activity(timeframe="1w") + +# Group changes by type +from collections import defaultdict +changes_by_type = defaultdict(list) +for change in weekly.changes: + changes_by_type[change.activity_type].append(change) + +# Show distribution +for type_, changes in changes_by_type.items(): + print(f"{type_}: {len(changes)} changes") +""" + } + ], + output_schema={ + "description": "Complete activity report showing recent changes and summary statistics", + "properties": { + "timeframe": { + "title": "Timeframe", + "type": "string", + "description": "Time period the activity covers (e.g. 1h, 1d, 1w, 1m)" + }, + "changes": { + "title": "Changes", + "type": "array", + "description": "List of individual changes in chronological order", + "items": { + "$ref": "#/definitions/ActivityChange" + } + }, + "summary": { + "$ref": "#/definitions/ActivitySummary", + "description": "Aggregated statistics about changes" + } + }, + "definitions": { + "ActivityChange": { + "description": "Detailed record of a single change in the system", + "properties": { + "activity_type": { + "type": "string", + "enum": ["document", "entity", "relation"], + "description": "Category of item that changed" + }, + "change_type": { + "type": "string", + "enum": ["created", "updated", "deleted"], + "description": "Type of change that occurred" + }, + "timestamp": { + "type": "string", + "format": "date-time", + "description": "When the change happened (ISO format)" + }, + "path_id": { + "type": "string", + "description": "Identifier for the changed item" + }, + "summary": { + "type": "string", + "description": "Human-readable description of the change" + }, + "content": { + "type": "string", + "description": "Optional details about the change", + "nullable": True + } + }, + "required": ["activity_type", "change_type", "timestamp", "path_id", "summary"] + }, + "ActivitySummary": { + "description": "Statistical overview of activity in the timeframe", + "properties": { + "document_changes": { + "type": "integer", + "description": "Number of document modifications", + "default": 0 + }, + "entity_changes": { + "type": "integer", + "description": "Number of entity modifications", + "default": 0 + }, + "relation_changes": { + "type": "integer", + "description": "Number of relationship changes", + "default": 0 + }, + "most_active_paths": { + "type": "array", + "items": {"type": "string"}, + "description": "List of paths with most changes" + } + } + } + } + } +) async def get_recent_activity( timeframe: str = "1d", activity_types: Optional[List[ActivityType]] = None, ) -> RecentActivity: """ Get recent activity across your knowledge base. - - Shows you what has changed recently including: - - Document changes - - Entity updates - - Relation modifications - - You can filter by: - - Timeframe (e.g., 1h, 1d, 1w, 1m) - - Activity types (document, entity, relation) - - Examples: - # Get all activity in last day - activity = await get_recent_activity() - - # Get only document changes - docs = await get_recent_activity( - timeframe="1h", - activity_types=[ActivityType.DOCUMENT] - ) - - Returns: - RecentActivity object with changes and summary """ - logger.debug(f"Getting recent activity (timeframe={timeframe}, " f"types={activity_types})") + logger.debug(f"Getting recent activity (timeframe={timeframe}, types={activity_types})") - # Build params + # Build params params = { "timeframe": timeframe, } diff --git a/src/basic_memory/mcp/tools/discovery.py b/src/basic_memory/mcp/tools/discovery.py index 0192d425..024d6c25 100644 --- a/src/basic_memory/mcp/tools/discovery.py +++ b/src/basic_memory/mcp/tools/discovery.py @@ -9,70 +9,218 @@ from basic_memory.schemas import EntityTypeList, ObservationCategoryList, TypedE from basic_memory.mcp.async_client import client -@mcp.tool() +@mcp.tool( + description=""" + List all unique entity types in use across the knowledge graph. + + This tool helps understand the structure of your knowledge base by showing: + - All entity types currently in use + - Custom types you've created + - System-defined types + + Useful for: + - Understanding knowledge organization + - Finding available entity types + - Discovering custom types + - Planning knowledge structure + """, + examples=[ + { + "name": "List Entity Types", + "description": "Show all entity types with counts", + "code": """ +# Get all entity types +types = await get_entity_types() + +# Count entities of each type +for entity_type in types: + entities = await list_by_type(entity_type) + print(f"{entity_type}: {len(entities.entities)} entities") +""" + }, + { + "name": "Find Custom Types", + "description": "Identify custom entity types", + "code": """ +# Get all types +types = await get_entity_types() + +# Separate system and custom types +system_types = {"component", "document", "feature", "test"} +custom_types = [t for t in types if t not in system_types] + +print("Custom entity types:") +for t in custom_types: + print(f"- {t}") +""" + } + ], + output_schema={ + "type": "array", + "description": "List of entity type strings", + "items": { + "type": "string", + "description": "Unique entity type identifier" + } + } +) async def get_entity_types() -> List[str]: - """List all unique entity types in use across the knowledge graph. - - Examples: - types = await get_entity_types() - - # Returns list of strings like: - # [ - # "technical_component", - # "specification", - # "decision", - # "feature" - # ] - - Returns: - List of unique entity type strings used in the knowledge graph - """ + """List all unique entity types in use.""" logger.debug("Getting all entity types") url = "/discovery/entity-types" response = await client.get(url) return EntityTypeList.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + List all unique observation categories used in the knowledge graph. + + Categories help organize different types of observations like: + - Technical details (tech) + - Design decisions (design) + - Features (feature) + - General notes (note) + - Issues/bugs (issue) + - Todo items (todo) + + This helps understand how knowledge is categorized and find + specific types of information. + """, + examples=[ + { + "name": "List Categories", + "description": "Show all observation categories", + "code": """ +# Get categories +categories = await get_observation_categories() + +# Group some recent entities by category +results = await search_nodes( + request=SearchNodesRequest( + query="database", + category=None # Search all categories + ) +) + +# Show observations by category +for category in categories: + obs = [o for e in results.matches + for o in e.observations + if o.category == category] + if obs: + print(f"\\n{category.upper()}:") + for o in obs: + print(f"- {o.content}") +""" + } + ], + output_schema={ + "type": "array", + "description": "List of observation category strings", + "items": { + "type": "string", + "enum": ["tech", "design", "feature", "note", "issue", "todo"], + "description": "Category identifier" + } + } +) async def get_observation_categories() -> List[str]: - """List all unique observation categories in use across the knowledge graph. - - Examples: - categories = await get_observation_categories() - - # Returns list of strings like: - # [ - # "tech", - # "design", - # "feature", - # "note" - # ] - - Returns: - List of unique observation category strings used in the knowledge graph - """ + """List all unique observation categories in use.""" logger.debug("Getting all observation categories") url = "/discovery/observation-categories" response = await client.get(url) return ObservationCategoryList.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + List all entities of a specific type with optional related entities. + + This tool provides: + - All entities of a given type + - Optional related entities + - Sorting options + - Complete entity information + + Useful for: + - Exploring entity collections + - Finding related entities + - Understanding entity relationships + - Analyzing knowledge structure + """, + examples=[ + { + "name": "List Components", + "description": "Show all technical components", + "code": """ +# Get components with relations +components = await list_by_type( + entity_type="component", + include_related=True +) + +# Show component dependencies +for entity in components.entities: + print(f"\\n{entity.name}") + deps = [r for r in entity.relations + if r.relation_type == "depends_on"] + if deps: + print("Dependencies:") + for dep in deps: + print(f"- {dep.to_id}") +""" + }, + { + "name": "Recent Features", + "description": "List recently updated features", + "code": """ +# Get features sorted by update time +features = await list_by_type( + entity_type="feature", + sort_by="updated_at" +) + +# Show recent features with status +print("Recent features:") +for entity in features.entities: + status = next((o.content for o in entity.observations + if o.category == "note"), "No status") + print(f"- {entity.name}: {status}") +""" + } + ], + output_schema={ + "description": "List of entities of a specific type", + "properties": { + "entity_type": { + "type": "string", + "description": "The type of entities listed" + }, + "entities": { + "type": "array", + "description": "List of matching entities", + "items": { + "$ref": "#/definitions/EntityResponse" + } + }, + "total": { + "type": "integer", + "description": "Total number of entities of this type" + }, + "include_related": { + "type": "boolean", + "description": "Whether related entities are included" + } + } + } +) async def list_by_type( - entity_type: str, include_related: bool = False, sort_by: Optional[str] = "updated_at" + entity_type: str, + include_related: bool = False, + sort_by: Optional[str] = "updated_at" ) -> TypedEntityList: - """List all entities of a specific type. - - Example: - # Get all features - features = await list_by_type("feature") - - # Get components with relations - components = await list_by_type( - "component", - include_related=True - ) - """ + """List all entities of a specific type.""" logger.debug(f"Listing entities of type: {entity_type}") params = {"include_related": "true" if include_related else "false"} if sort_by: diff --git a/src/basic_memory/mcp/tools/documents.py b/src/basic_memory/mcp/tools/documents.py index da331f21..79e9bb55 100644 --- a/src/basic_memory/mcp/tools/documents.py +++ b/src/basic_memory/mcp/tools/documents.py @@ -8,156 +8,360 @@ from basic_memory.schemas.response import DocumentResponse, DocumentCreateRespon from basic_memory.mcp.async_client import client -@mcp.tool() -async def create_document(request: DocumentRequest) -> DocumentCreateResponse: - """Create a new markdown document. +@mcp.tool( + description=""" + Create a new markdown document in the knowledge base. - Examples: - # Create a technical specification - request = DocumentRequest( - path="specs/memory_format.md", - content='''# Memory Format Specification + This tool stores markdown documents with: + - Structured frontmatter metadata + - Rich markdown content + - Version tracking via checksums + - Automatic timestamp management + - Optional custom metadata - ## Overview - This document defines the standard format for memory files. + Documents are stored in a git-friendly format and can be + edited either through the API or directly in the filesystem. + """, + examples=[ + { + "name": "Create Technical Spec", + "description": "Create a new technical specification document", + "code": """ +# Create new spec with metadata +spec = await create_document( + request=DocumentRequest( + path_id="specs/memory_format.md", + content='''# Memory Format Specification - ## Format - - Markdown with frontmatter - - UTF-8 encoding - - Required metadata fields - ''', - doc_metadata={ - "author": "AI team", - "status": "draft", - "version": "0.1" +## Overview +This document defines our standard format. + +## Structure +1. Frontmatter for metadata +2. Markdown content for documentation +3. Optional structured data sections''', + doc_metadata={ + "status": "draft", + "version": "0.1", + "reviewers": ["@alice", "@bob"] + } + ) +) + +print(f"Created: {spec.path_id}") +print(f"Version: {spec.doc_metadata['version']}") +""" + }, + { + "name": "Create Design Document", + "description": "Document a design decision with context", + "code": """ +# Create design document +design = await create_document( + request=DocumentRequest( + path_id="design/database_schema.md", + content='''# Database Schema Design + +## Decision +Using SQLite for local-first storage. + +## Context +Need reliable local storage with SQL features. + +## Consequences ++ Simple deployment ++ Local-first operation +- Limited concurrent access''', + doc_metadata={ + "type": "decision", + "status": "accepted", + "date": "2024-12-25" + } + ) +) +""" + } + ], + output_schema={ + "description": "Created document information", + "properties": { + "path_id": { + "type": "string", + "description": "Document path and filename" + }, + "checksum": { + "type": "string", + "description": "Content checksum for version tracking" + }, + "doc_metadata": { + "type": "object", + "description": "Custom document metadata", + "additionalProperties": True + }, + "created_at": { + "type": "string", + "format": "date-time", + "description": "Creation timestamp" + }, + "updated_at": { + "type": "string", + "format": "date-time", + "description": "Last modification timestamp" } - ) - response = await create_document(request) - - # Response contains document info: - # DocumentCreateResponse( - # path="specs/memory_format.md", - # checksum="abc123...", - # doc_metadata={...}, - # created_at="2024-12-25T12:00:00Z", - # updated_at="2024-12-25T12:00:00Z" - # ) - """ + }, + "required": ["path_id", "checksum", "created_at", "updated_at"] + } +) +async def create_document(request: DocumentRequest) -> DocumentCreateResponse: + """Create a new markdown document.""" url = "/documents/create" response = await client.post(url, json=request.model_dump()) return DocumentCreateResponse.model_validate(response.json()) -@mcp.tool() -async def update_document(request: DocumentRequest) -> DocumentResponse: - """Update an existing document. +@mcp.tool( + description=""" + Update an existing document while preserving its history. + + This tool handles: + - Content updates + - Metadata changes + - Version tracking + - Timestamp management + + The update preserves document history and maintains + consistency with any linked knowledge graph entities. + """, + examples=[ + { + "name": "Update Content", + "description": "Add new content to existing document", + "code": """ +# Update implementation details +updated = await update_document( + request=DocumentRequest( + path_id="docs/implementation.md", + content='''# Implementation Details - Examples: - # Update implementation docs with new details - request = DocumentRequest( - path="docs/implementation.md", - content='''# Implementation Details +## Recent Updates +- Added async support +- Improved error handling +- Enhanced performance - ## Recent Changes - - Added FTS5 support - - Improved error handling - - Enhanced sync reliability - ''', - doc_metadata={ - "last_reviewed": "2024-12-25", - "status": "current" +## New Features +- Batch processing +- Automatic retries +- Error recovery''', + doc_metadata={ + "status": "current", + "last_updated": "2024-12-25" + } + ) +) + +print(f"Updated: {updated.path_id}") +print(f"New checksum: {updated.checksum}") +""" + } + ], + output_schema={ + "description": "Updated document with content", + "properties": { + "path_id": { + "type": "string", + "description": "Document path and filename" + }, + "content": { + "type": "string", + "description": "Current document content" + }, + "checksum": { + "type": "string", + "description": "New content checksum" + }, + "doc_metadata": { + "type": "object", + "description": "Current document metadata" + }, + "created_at": { + "type": "string", + "format": "date-time" + }, + "updated_at": { + "type": "string", + "format": "date-time" } - ) - response = await update_document(request) - - # Response contains updated document: - # DocumentResponse( - # path="docs/implementation.md", - # content="# Implementation Details\n...", - # checksum="def456...", - # doc_metadata={...}, - # created_at="2024-12-20T10:00:00Z", - # updated_at="2024-12-25T14:30:00Z" - # ) - """ + }, + "required": ["path_id", "content", "checksum"] + } +) +async def update_document(request: DocumentRequest) -> DocumentResponse: + """Update an existing document.""" url = f"/documents/{request.path_id}" response = await client.put(url, json=request.model_dump()) return DocumentResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + Retrieve a document's content and metadata. + + This tool provides access to: + - Full document content + - Current metadata + - Version information + - Timestamps + + Documents are returned with their complete context, useful + for reading or preparing updates. + """, + examples=[ + { + "name": "Read Documentation", + "description": "Load and display a document", + "code": """ +# Get API documentation +doc = await get_document("docs/api_reference.md") + +# Show document info +print(f"Document: {doc.path_id}") +print(f"Status: {doc.doc_metadata.get('status', 'unknown')}") +print("\\nContent:") +print(doc.content) +""" + } + ], + output_schema={ + "description": "Complete document information", + "properties": { + "path_id": { + "type": "string", + "description": "Document identifier" + }, + "content": { + "type": "string", + "description": "Document content" + }, + "checksum": { + "type": "string", + "description": "Content checksum" + }, + "doc_metadata": { + "type": "object", + "description": "Document metadata" + } + } + } +) async def get_document(path: DocumentPathId) -> DocumentResponse: - """Get a document by its path. - - Examples: - # Load an API specification - response = await get_document("specs/api_format.md") - - # Response contains complete document: - # DocumentResponse( - # path="specs/api_format.md", - # content="# API Format\n\n## Endpoints\n...", - # checksum="789ghi...", - # doc_metadata={ - # "status": "current", - # "version": "1.0" - # }, - # created_at="2024-12-01T09:00:00Z", - # updated_at="2024-12-20T15:45:00Z" - # ) - - # Load implementation details - response = await get_document("docs/implementation.md") - """ + """Get a document by its path.""" url = f"/documents/{path}" response = await client.get(url) return DocumentResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + List all documents in the knowledge base. + + Provides an overview of the document collection including: + - Document paths and names + - Metadata for each document + - Version information + - Timestamps + + Useful for browsing content or finding specific documents. + """, + examples=[ + { + "name": "List All Documents", + "description": "Show overview of all documents", + "code": """ +# Get document listing +docs = await list_documents() + +# Group by status +from collections import defaultdict +by_status = defaultdict(list) + +for doc in docs: + status = doc.doc_metadata.get('status', 'unknown') + by_status[status].append(doc) + +# Show summary +for status, items in by_status.items(): + print(f"\\n{status.title()} Documents:") + for doc in items: + print(f"- {doc.path_id}") +""" + } + ], + output_schema={ + "description": "List of document information", + "type": "array", + "items": { + "type": "object", + "properties": { + "path_id": { + "type": "string", + "description": "Document path" + }, + "checksum": { + "type": "string", + "description": "Version checksum" + }, + "doc_metadata": { + "type": "object", + "description": "Document metadata" + } + } + } + } +) async def list_documents() -> List[DocumentCreateResponse]: - """List all documents in the system. - - Examples: - # Get all documents with metadata - documents = await list_documents() - - # Response is list of document info: - # [ - # DocumentCreateResponse( - # path="specs/format.md", - # checksum="abc123...", - # doc_metadata={"status": "draft"}, - # created_at="2024-12-01T09:00:00Z", - # updated_at="2024-12-25T10:30:00Z" - # ), - # DocumentCreateResponse( - # path="docs/implementation.md", - # checksum="def456...", - # doc_metadata={"status": "current"}, - # created_at="2024-12-20T10:00:00Z", - # updated_at="2024-12-25T14:30:00Z" - # ) - # ] - """ + """List all documents in the system.""" url = "/documents/list" response = await client.get(url) return [DocumentCreateResponse.model_validate(doc) for doc in response.json()] -@mcp.tool() +@mcp.tool( + description=""" + Delete a document from the knowledge base. + + This tool: + - Removes the document file + - Updates related indexes + - Maintains consistency + + Note that deletion is permanent and cannot be undone + through the API (though git history may preserve it). + """, + examples=[ + { + "name": "Remove Document", + "description": "Delete an obsolete document", + "code": """ +# Delete old specification +result = await delete_document("specs/old_format.md") +if result['deleted']: + print("Document successfully removed") +""" + } + ], + output_schema={ + "description": "Deletion result", + "type": "object", + "properties": { + "deleted": { + "type": "boolean", + "description": "Whether deletion succeeded" + } + } + } +) async def delete_document(path: DocumentPathId) -> Dict[str, bool]: - """Delete a document. - - Examples: - # Remove an obsolete document - result = await delete_document("docs/outdated_spec.md") - - # Response indicates success: - # { - # "deleted": true - # } - """ + """Delete a document.""" url = f"/documents/{path}" response = await client.delete(url) if response.status_code == 204: diff --git a/src/basic_memory/mcp/tools/help.py b/src/basic_memory/mcp/tools/help.py index 8ff2c6e0..44cbc6ef 100644 --- a/src/basic_memory/mcp/tools/help.py +++ b/src/basic_memory/mcp/tools/help.py @@ -7,24 +7,132 @@ from basic_memory.mcp.server import mcp @mcp.tool( category="system", + description=""" + Get schema information about available tools. + + This tool provides access to the MCP schema catalog, showing: + - Available tools and their capabilities + - Input/output type definitions + - Example usage patterns + - Related schema models + + You can: + - Get the full tool catalog + - Look up specific tools + - Control example inclusion + - Access referenced models + + The schema information helps understand tool capabilities + and ensure correct usage. + """, examples=[ { - "name": "Get All Tools", - "description": "Get complete schema catalog for all tools", - "code": "catalog = await get_schema()", + "name": "View All Tools", + "description": "Get complete schema catalog", + "code": """ +# Get full tool catalog +catalog = await get_schema() + +# Show available tools by category +for category, info in catalog['categories'].items(): + print(f"\\n{category.title()}:") + for tool in info['tools']: + print(f"- {tool}") +""" }, { - "name": "Get Specific Tool", - "description": "Get schema for a specific tool", - "code": 'tool_schema = await get_schema("create_entity")', + "name": "Tool Details", + "description": "Examine specific tool schema", + "code": """ +# Get schema for create_entities +tool = await get_schema( + tool_name="create_entities", + include_referenced=True # Include type definitions +) + +# Show input/output types +print("Inputs:") +for param, info in tool['tools']['create_entities']['inputSchema']['properties'].items(): + print(f"- {param}: {info.get('description', '')}") + +print("\\nOutput:") +print(tool['tools']['create_entities']['outputSchema']['description']) +""" }, + { + "name": "Simple Schema", + "description": "Get minimal schema without examples", + "code": """ +# Get core schema without examples +schema = await get_schema( + include_examples=False, + include_referenced=False +) + +# List available tools +tools = list(schema['tools'].keys()) +print("Available tools:") +for tool in sorted(tools): + print(f"- {tool}") +""" + } ], + output_schema={ + "description": "Tool schema catalog", + "properties": { + "tools": { + "type": "object", + "description": "Map of tool names to their schemas", + "additionalProperties": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "description": {"type": "string"}, + "category": {"type": "string"}, + "inputSchema": {"type": "object"}, + "outputSchema": {"type": "object"}, + "examples": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "description": {"type": "string"}, + "code": {"type": "string"} + } + } + } + } + } + }, + "categories": { + "type": "object", + "description": "Tool categories and their tools", + "additionalProperties": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "tools": { + "type": "array", + "items": {"type": "string"} + } + } + } + }, + "referencedModels": { + "type": "object", + "description": "Shared type definitions", + "additionalProperties": {"type": "object"} + } + } + } ) async def get_schema( - tool_name: Optional[str] = None, include_examples: bool = True, include_referenced: bool = True + tool_name: Optional[str] = None, + include_examples: bool = True, + include_referenced: bool = True ) -> Dict: - """Get schema information about available tools. - """ + """Get schema information about available tools.""" # Our tool manager has the enhanced schema support catalog = mcp._tool_manager.get_schema_catalog() @@ -41,7 +149,7 @@ async def get_schema( if include_referenced: return { "tools": {tool_name: tool_schema}, - "referencedModels": tool_schema.get("referencedModels", {}), + "referencedModels": tool_schema.get("referencedModels", {}) } else: return {"tools": {tool_name: tool_schema}} @@ -53,4 +161,4 @@ async def get_schema( for tool in result["tools"].values(): tool.pop("examples", None) - return result + return result \ No newline at end of file diff --git a/src/basic_memory/mcp/tools/knowledge.py b/src/basic_memory/mcp/tools/knowledge.py index 894dcb7c..35ebf66e 100644 --- a/src/basic_memory/mcp/tools/knowledge.py +++ b/src/basic_memory/mcp/tools/knowledge.py @@ -23,11 +23,29 @@ from basic_memory.services.exceptions import EntityNotFoundError @mcp.tool( category="knowledge", + description=""" + Create new entities in the knowledge graph. + + Entities are the core building blocks of the knowledge graph. Each entity: + - Has a unique name and type + - Can have multiple observations + - Can have relations to other entities + - Maintains creation/update timestamps + - Supports optional descriptions + + Entity types help organize knowledge and enable patterns like: + - Components for technical implementations + - Features for user-facing capabilities + - Concepts for abstract ideas + - Decisions for architectural choices + - Documents for detailed writeups + """, examples=[ { "name": "Create Component", "description": "Create a new technical component", "code": """ +# Create search service component await create_entities({ "entities": [{ "name": "SearchService", @@ -35,20 +53,36 @@ await create_entities({ "description": "Full-text search capability", "observations": [ "Implements FTS5 for better performance", - "Supports fuzzy matching" + "Supports fuzzy matching", + "Handles multiple indexes" ] }] }) -""", +""" + }, + { + "name": "Create Feature", + "description": "Document a user-facing feature", + "code": """ +# Create feature with implementation notes +await create_entities({ + "entities": [{ + "name": "SemanticSearch", + "entity_type": "feature", + "description": "Natural language search across knowledge base", + "observations": [ + "Uses embeddings for matching", + "Supports fuzzy queries", + "Ranks results by relevance" + ] + }] +}) +""" } - ], + ] ) async def create_entities(request: CreateEntityRequest) -> EntityListResponse: - """Create new entities in the knowledge graph. - - Entities can include initial observations and properties. Entity IDs - are automatically generated from the type and name. - """ + """Create new entities in the knowledge graph.""" url = "/knowledge/entities" response = await client.post(url, json=request.model_dump()) return EntityListResponse.model_validate(response.json()) @@ -56,11 +90,27 @@ async def create_entities(request: CreateEntityRequest) -> EntityListResponse: @mcp.tool( category="knowledge", + description=""" + Create relations between existing entities. + + Relations form the edges of the knowledge graph, connecting entities with: + - Directional relationships (from_id -> to_id) + - Typed connections (implements, depends_on, etc.) + - Optional context notes + - Automatic timestamp tracking + + Common relation patterns: + - Component implements Feature + - Component depends_on Component + - Test validates Component + - Document describes Feature + """, examples=[ { "name": "Add Dependency", "description": "Create dependency relationship between components", "code": """ +# Document component dependency await create_relations({ "relations": [{ "from_id": "component/search_service", @@ -69,9 +119,24 @@ await create_relations({ "context": "Needs storage for search indexes" }] }) -""", +""" + }, + { + "name": "Link Implementation", + "description": "Connect implementation to feature", + "code": """ +# Link component to feature +await create_relations({ + "relations": [{ + "from_id": "component/search_service", + "to_id": "feature/semantic_search", + "relation_type": "implements", + "context": "Primary search implementation" + }] +}) +""" } - ], + ] ) async def create_relations(request: CreateRelationsRequest) -> EntityListResponse: """Create relations between existing entities.""" @@ -82,57 +147,48 @@ async def create_relations(request: CreateRelationsRequest) -> EntityListRespons @mcp.tool( category="knowledge", + description=""" + Get complete information about a specific entity. + + Returns the full entity context including: + - Basic entity details (name, type, description) + - All observations with categories + - All relations (both incoming and outgoing) + - Timestamps and metadata + + Useful for: + - Understanding entity details + - Following relationships + - Finding related knowledge + - Analyzing implementation patterns + """, examples=[ { - "name": "Get Entity Details", - "description": "Load complete entity information", + "name": "View Component Details", + "description": "Get complete component information", "code": """ -# Get component details -entity = await get_entity("component/search_service") -print(f"Name: {entity.name}") -print(f"Type: {entity.entity_type}") -for obs in entity.observations: - print(f"- {obs.content}") -""", +# Get component implementation details +component = await get_entity("component/search_service") + +# Show technical details +tech_notes = [obs for obs in component.observations + if obs.category == "tech"] +print(f"{component.name} Implementation:") +for note in tech_notes: + print(f"- {note.content}") + +# Show dependencies +deps = [rel for rel in component.relations + if rel.relation_type == "depends_on"] +print("\\nDependencies:") +for dep in deps: + print(f"- {dep.to_id}") +""" } - ], + ] ) async def get_entity(path_id: PathId) -> EntityResponse: - """Get a specific entity by its path_id. - - Examples: - # Load implementation details - response = await get_entity("component/memory_service") - - # Response contains complete entity: - # EntityResponse( - # path_id="component/memory_service", - # name="memory_service", - # entity_type="component", - # description="Core knowledge persistence service", - # observations=[ - # Observation( - # category="TECH", - # content="Using SQLite for storage", - # context="Initial implementation" - # ), - # ... - # ], - # relations=[ - # Relation( - # from_id="component/memory_service", - # to_id="component/file_service", - # relation_type="depends_on" - # ), - # ... - # ] - # ) - - # Load and analyze a design spec - spec = await get_entity("specification/file_format") - decisions = [obs for obs in spec.observations - if obs.category == ObservationCategory.DESIGN] - """ + """Get a specific entity by its path_id.""" try: url = f"/knowledge/entities/{path_id}" response = await client.get(url) @@ -141,136 +197,181 @@ async def get_entity(path_id: PathId) -> EntityResponse: response.raise_for_status() return EntityResponse.model_validate(response.json()) except httpx.HTTPStatusError as e: - # If we got a 404, the entity doesn't exist if e.response.status_code == 404: raise EntityNotFoundError(f"Entity not found: {path_id}") - # For any other HTTP error, re-raise raise -@mcp.tool() +@mcp.tool( + description=""" + Add new observations to an existing entity. + + Observations capture atomic pieces of knowledge about an entity: + - Technical details + - Design decisions + - Feature specifications + - Implementation notes + - Issues or concerns + - Todo items + + Each observation has: + - A category for organization + - Content describing the observation + - Optional context for additional detail + - Automatic timestamp tracking + """, + examples=[ + { + "name": "Add Implementation Notes", + "description": "Document technical implementation details", + "code": """ +# Add technical observations +await add_observations( + request=AddObservationsRequest( + path_id="component/search_service", + context="Performance optimization", + observations=[ + ObservationCreate( + category="tech", + content="Implemented FTS5 for better search" + ), + ObservationCreate( + category="tech", + content="Added result caching" + ), + ObservationCreate( + category="design", + content="Chose FTS5 for better ranking" + ) + ] + ) +) +""" + } + ] +) async def add_observations(request: AddObservationsRequest) -> EntityResponse: - """Add observations to an existing entity. - - Examples: - # Document implementation decisions with context - request = AddObservationsRequest( - path_id="component/search_service", - context="Performance optimization meeting", - observations=[ - ObservationCreate( - category=ObservationCategory.TECH, - content="Implementing FTS5 for full-text search" - ), - ObservationCreate( - category=ObservationCategory.DESIGN, - content="Chose FTS5 for better ranking and phrase queries" - ), - ObservationCreate( - category=ObservationCategory.FEATURE, - content="Added support for fuzzy matching" - ) - ] - ) - response = await add_observations(request) - - # Response shows entity with new observations: - # EntityResponse( - # path_id="component/search_service", - # observations=[ - # Observation( - # category="TECH", - # content="Implementing FTS5 for full-text search", - # context="Performance optimization meeting" - # ), - # ... - # ] - # ) - """ + """Add observations to an existing entity.""" url = "/knowledge/observations" response = await client.post(url, json=request.model_dump()) return EntityResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + Delete specific observations from an entity. + + This tool: + - Removes selected observations + - Maintains entity history + - Updates timestamps + - Preserves relations + + Observations must match exactly for deletion. + The operation is selective - only specified + observations are removed. + """, + examples=[ + { + "name": "Remove Obsolete Notes", + "description": "Delete outdated observations", + "code": """ +# Remove old implementation notes +await delete_observations( + request=DeleteObservationsRequest( + path_id="component/indexer", + observations=[ + "Using old indexing algorithm", + "Temporary workaround for issue #123" + ] + ) +) +""" + } + ] +) async def delete_observations(request: DeleteObservationsRequest) -> EntityResponse: - """Delete specific observations from an entity. - - Examples: - # Remove obsolete implementation notes - request = DeleteObservationsRequest( - path_id="component/indexer", - observations=[ - "Using old indexing algorithm", - "Temporary workaround for issue #123" - ] - ) - response = await delete_observations(request) - - # Response shows entity with observations removed: - # EntityResponse( - # path_id="component/indexer", - # observations=[...] # Remaining observations - # ) - """ + """Delete specific observations from an entity.""" url = "/knowledge/observations/delete" response = await client.post(url, json=request.model_dump()) return EntityResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + Delete relations between entities. + + This tool: + - Removes specific relationships + - Updates both source and target entities + - Maintains entity history + - Preserves observations + + Relations must match exactly (from_id, to_id, and type) + for deletion. The operation only affects the specified + relations, leaving other connections intact. + """, + examples=[ + { + "name": "Remove Dependency", + "description": "Delete an obsolete dependency", + "code": """ +# Remove old dependency +await delete_relations( + request=DeleteRelationsRequest( + relations=[{ + "from_id": "component/search", + "to_id": "component/old_index", + "relation_type": "depends_on" + }] + ) +) +""" + } + ] +) async def delete_relations(request: DeleteRelationsRequest) -> EntityListResponse: - """Delete relations between entities. - - Examples: - # Remove obsolete dependency - request = DeleteRelationsRequest( - relations=[ - Relation( - from_id="component/search", - to_id="component/old_index", - relation_type="depends_on" - ) - ] - ) - response = await delete_relations(request) - - # Response shows updated entities: - # EntityListResponse( - # entities=[ - # EntityResponse( # search component - # relations=[...] # Remaining relations - # ), - # EntityResponse( # old_index component - # relations=[...] # Remaining relations - # ) - # ] - # ) - """ + """Delete relations between entities.""" url = "/knowledge/relations/delete" response = await client.post(url, json=request.model_dump()) return EntityListResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + Delete entities from the knowledge graph. + + This operation: + 1. Removes the entity completely + 2. Deletes all its observations + 3. Removes all relations (both ways) + 4. Updates related indexes + + This is a permanent operation that cannot be + undone through the API. Use with caution. + """, + examples=[ + { + "name": "Remove Old Components", + "description": "Delete obsolete components", + "code": """ +# Remove deprecated components +await delete_entities( + request=DeleteEntitiesRequest( + path_ids=[ + "component/old_service", + "test/obsolete_test" + ] + ) +) +""" + } + ] +) async def delete_entities(request: DeleteEntitiesRequest) -> Dict[str, bool]: - """Delete entities from the knowledge graph. - - Examples: - # Remove obsolete components - request = DeleteEntitiesRequest( - path_ids=[ - "component/old_service", - "test/obsolete_test" - ] - ) - response = await delete_entities(request) - - # Response indicates success: - # { - # "deleted": true - # } - """ + """Delete entities from the knowledge graph.""" url = "/knowledge/entities/delete" response = await client.post(url, json=request.model_dump()) + if response.status_code == 204: + return {"deleted": True} return response.json() diff --git a/src/basic_memory/mcp/tools/search.py b/src/basic_memory/mcp/tools/search.py index 344282a6..bfc46fc1 100644 --- a/src/basic_memory/mcp/tools/search.py +++ b/src/basic_memory/mcp/tools/search.py @@ -8,81 +8,232 @@ from basic_memory.schemas.response import SearchNodesResponse, EntityResponse from basic_memory.mcp.async_client import client -@mcp.tool() +@mcp.tool( + description=""" + Search for entities in the knowledge graph. + + This is a powerful semantic search that looks across: + - Entity names and types + - Descriptions and metadata + - Observation content + - Relation contexts + + Features: + - Case-insensitive matching + - Partial word matches + - Category filtering + - Returns full entity context + - Natural language friendly + + The search combines multiple approaches to find relevant entities, + including text matching, category filtering, and context awareness. + Results include complete entity information with observations + and relations to help understand the context. + """, + examples=[ + { + "name": "Basic Text Search", + "description": "Simple search across all content", + "code": """ +# Search for SQLite-related entities +results = await search_nodes( + request=SearchNodesRequest(query="sqlite database") +) + +# Show matches with context +for entity in results.matches: + print(f"\\n{entity.entity_type}: {entity.name}") + print(f"Description: {entity.description}") + print("Relevant observations:") + for obs in entity.observations: + print(f"- {obs.content}") +""" + }, + { + "name": "Category-Filtered Search", + "description": "Find technical implementation details", + "code": """ +# Search for tech implementation details +tech_results = await search_nodes( + request=SearchNodesRequest( + query="async implementation", + category="tech" # Only tech observations + ) +) + +# Show technical findings +for entity in tech_results.matches: + tech_obs = [o for o in entity.observations + if o.category == "tech"] + print(f"\\n{entity.name} - {len(tech_obs)} tech notes") + for obs in tech_obs: + print(f"- {obs.content}") +""" + }, + { + "name": "Design Decision Search", + "description": "Find architectural decisions", + "code": """ +# Search for design decisions +design = await search_nodes( + request=SearchNodesRequest( + query="architecture pattern decision", + category="design" # Only design observations + ) +) + +# Show decision history +for entity in design.matches: + print(f"\\n{entity.name}") + for obs in entity.observations: + if obs.context: + print(f"{obs.context}:") + print(f"- {obs.content}") +""" + } + ], + output_schema={ + "description": "Search results with matching entities and query info", + "properties": { + "matches": { + "type": "array", + "description": "List of entities matching the search criteria", + "items": { + "$ref": "#/definitions/EntityResponse" + } + }, + "query": { + "type": "string", + "description": "Original search query for reference" + } + }, + "definitions": { + "EntityResponse": { + "description": "Complete entity information", + "properties": { + "path_id": { + "type": "string", + "description": "Unique identifier for the entity" + }, + "name": { + "type": "string", + "description": "Human-readable entity name" + }, + "entity_type": { + "type": "string", + "description": "Classification of the entity" + }, + "description": { + "type": "string", + "description": "Overview of the entity's purpose", + "nullable": True + }, + "observations": { + "type": "array", + "description": "List of observations about the entity", + "items": { + "$ref": "#/definitions/ObservationResponse" + } + }, + "relations": { + "type": "array", + "description": "List of relationships with other entities", + "items": { + "$ref": "#/definitions/RelationResponse" + } + } + }, + "required": ["path_id", "name", "entity_type"] + } + } + } +) async def search_nodes(request: SearchNodesRequest) -> SearchNodesResponse: - """Search for entities in the knowledge graph. - - Examples: - # Find technical implementation details - request = SearchNodesRequest( - query="SQLite implementation", - category=ObservationCategory.TECH - ) - response = await search_nodes(request) - - # Response contains matching entities: - # SearchNodesResponse( - # matches=[ - # EntityResponse( # First matching entity - # path_id="component/memory_service", - # name="memory_service", - # description="Core service for persistence", - # observations=[ - # Observation( - # category="TECH", - # content="Using SQLite for storage" - # ) - # ] - # ), - # EntityResponse(...) # Other matches - # ], - # query="SQLite implementation" - # ) - - # Find design decisions - request = SearchNodesRequest( - query="database design decision", - category=ObservationCategory.DESIGN - ) - response = await search_nodes(request) - """ + """Search for entities in the knowledge graph.""" url = "/knowledge/search" response = await client.post(url, json=request.model_dump()) return SearchNodesResponse.model_validate(response.json()) -@mcp.tool() +@mcp.tool( + description=""" + Load multiple entities by their path_ids. + + This tool efficiently loads multiple entities in a single request, + retrieving their complete information including observations + and relations. It's particularly useful for: + + - Following relation chains + - Loading related entities + - Batch entity retrieval + - Context building + + The response maps each path_id to its full entity data, + making it easy to access specific entities while maintaining + their relationships. + """, + examples=[ + { + "name": "Load Related Components", + "description": "Load a component and its dependencies", + "code": """ +# Load component and related specs +response = await open_nodes( + request=OpenNodesRequest( + path_ids=[ + "component/memory_service", + "component/file_service", + "specification/file_format" + ] + ) +) + +# Show component relationships +for path_id, entity in response.items(): + print(f"\\n{entity.name}") + print("Relations:") + for rel in entity.relations: + print(f"- {rel.relation_type} {rel.to_id}") +""" + }, + { + "name": "Feature Implementation Chain", + "description": "Load feature with implementation and tests", + "code": """ +# Load entire feature chain +chain = await open_nodes( + request=OpenNodesRequest( + path_ids=[ + "feature/search", # The feature + "component/search", # Implementation + "test/search_test", # Testing + "document/search_spec" # Documentation + ] + ) +) + +# Show implementation status +feature = chain["feature/search"] +impl = chain["component/search"] +test = chain["test/search_test"] + +print(f"Feature: {feature.name}") +print(f"Implementation: {impl.description}") +print(f"Test Status: {'test' in [r.relation_type for r in test.relations]}") +""" + } + ], + output_schema={ + "description": "Map of path_ids to their complete entity data", + "type": "object", + "additionalProperties": { + "$ref": "#/definitions/EntityResponse", + "description": "Full entity data including observations and relations" + } + } +) async def open_nodes(request: OpenNodesRequest) -> Dict[str, EntityResponse]: - """Load multiple entities by their path_ids. - - Examples: - # Load related components and their specs - request = OpenNodesRequest( - path_ids=[ - "component/memory_service", - "component/file_service", - "specification/file_format" - ] - ) - response = await open_nodes(request) - - # Response maps path_ids to entities: - # { - # "component/memory_service": EntityResponse(...), - # "component/file_service": EntityResponse(...), - # "specification/file_format": EntityResponse(...) - # } - - # Follow relation chains - request = OpenNodesRequest( - path_ids=[ - "feature/search", # The feature - "component/search_service", # Implementation - "test/search_integration" # Testing - ] - ) - response = await open_nodes(request) - """ + """Load multiple entities by their path_ids.""" url = "/knowledge/nodes" response = await client.post(url, json=request.model_dump()) return {