mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
update tool schema docs
This commit is contained in:
phernandez
@@ -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,
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user