refactor tools, remove BasicMemoryServer, update mcp sdk

src/basic_memory/api/routers/knowledge_router.py

@@ -1,6 +1,8 @@
 """Router for knowledge graph operations."""
 
-from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends
+from typing import Annotated
+
+from fastapi import APIRouter, HTTPException, BackgroundTasks, Depends, Query
 from loguru import logger
 
 from basic_memory.deps import (
@@ -8,7 +10,6 @@ from basic_memory.deps import (
     get_search_service,
     RelationServiceDep,
     ObservationServiceDep,
-    FileServiceDep,
 )
 from basic_memory.schemas import (
     CreateEntityRequest,
@@ -16,7 +17,6 @@ from basic_memory.schemas import (
     CreateRelationsRequest,
     EntityResponse,
     AddObservationsRequest,
-    OpenNodesRequest,
     DeleteEntitiesResponse,
     DeleteObservationsRequest,
     DeleteRelationsRequest,
@@ -119,9 +119,7 @@ async def add_observations(
 @router.get("/entities/{permalink:path}", response_model=EntityResponse)
 async def get_entity(
     entity_service: EntityServiceDep,
-    file_service: FileServiceDep,
     permalink: PathId,
-    content: bool = False,  # New parameter
 ) -> EntityResponse:
     """Get a specific entity by ID.
 
@@ -133,22 +131,19 @@ async def get_entity(
     try:
         entity = await entity_service.get_by_permalink(permalink)
         entity_response = EntityResponse.model_validate(entity)
-
-        if content:  # Load content if requested
-            content = await file_service.read_entity_content(entity)
-            entity_response.content = content
-
         return entity_response
     except EntityNotFoundError:
         raise HTTPException(status_code=404, detail=f"Entity with {permalink} not found")
 
 
-@router.post("/nodes", response_model=EntityListResponse)
-async def open_nodes(
-    data: OpenNodesRequest, entity_service: EntityServiceDep
+@router.get("/entities", response_model=EntityListResponse)
+async def get_entities(
+    entity_service: EntityServiceDep,
+    permalink: Annotated[list[str] | None, Query()] = None,
 ) -> EntityListResponse:
-    """Open specific nodes"""
-    entities = await entity_service.open_nodes(data.permalinks)
+    """Open specific entities"""
+    # permalink is a list of parameters on the request ?permalink=foo
+    entities = await entity_service.get_entities_by_permalinks(permalink)
     return EntityListResponse(
         entities=[EntityResponse.model_validate(entity) for entity in entities]
     )

src/basic_memory/mcp/async_client.py

@@ -6,3 +6,5 @@ BASE_URL = "http://test"
 
 # Create shared async client
 client = AsyncClient(transport=ASGITransport(app=fastapi_app), base_url=BASE_URL)
+
+

src/basic_memory/mcp/main.py

@@ -3,46 +3,21 @@
 Creates and configures the shared MCP instance and handles server startup.
 """
 
-import sys
 from loguru import logger
 
 from basic_memory.config import config
+
 # Import shared mcp instance
 from basic_memory.mcp.server import mcp
 
 # Import tools to register them
-from basic_memory.mcp.tools import knowledge, search, discovery, help
-__all__ = ["mcp", "knowledge", "search", "discovery", "help"]
+from basic_memory.mcp.tools import knowledge, search, discussion, activity
 
-
-def setup_logging(home_dir: str = config.home, log_file: str = "basic-memory.log"):
-    """Configure logging for the application."""
-    # Remove default handler
-    logger.remove()
-    log = f"{home_dir}/{log_file}"
-    
-    # Add file handler with rotation
-    logger.add(
-        log,
-        rotation="100 MB",
-        retention="10 days",
-        backtrace=True,
-        diagnose=True,
-        enqueue=True,
-        colorize=False,
-    )
-
-    # Add stderr handler
-    logger.add(
-        sys.stderr,
-        colorize=True,
-    )
+__all__ = ["knowledge", "search", "discussion", "activity"]
 
 
 if __name__ == "__main__":
-    
     home_dir = config.home
-    setup_logging(home_dir)
     logger.info("Starting Basic Memory MCP server")
-    logger.info(f"Home directory: {home_dir}" )
-    mcp.run()
+    logger.info(f"Home directory: {home_dir}")
+    mcp.run()

src/basic_memory/mcp/server.py

@@ -1,188 +1,9 @@
 """Enhanced FastMCP server instance for Basic Memory."""
 
-from typing import Any, Callable, Dict, List, Optional, Type, Union
-
-from fastmcp import FastMCP
-import inspect
-from pydantic import BaseModel, Field, TypeAdapter
-from loguru import logger
-
-from fastmcp.tools import Tool as FastMCPTool
-from fastmcp.tools.tool_manager import ToolManager as FastMCPToolManager
-
-
-class BasicMemoryServer(FastMCP):
-    """Enhanced FastMCP server with schema support."""
-
-    def __init__(self, name: str | None = None, **settings: Any):
-        super().__init__(name=name, **settings)
-        # Replace default tool manager with our enhanced version
-        self._tool_manager = EnhancedToolManager(
-            warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
-        )
-
-    def tool(
-        self,
-        name: Optional[str] = None,
-        description: Optional[str] = None,
-        examples: Optional[List[Dict]] = None,
-        category: Optional[str] = None,
-        output_model: Any = None,
-    ):
-        """Decorator to register an enhanced tool."""
-        def decorator(fn: Callable) -> Callable:
-            tool = self._tool_manager.add_tool(
-                fn, 
-                name=name, 
-                description=description, 
-                examples=examples, 
-                category=category,
-                output_model=output_model
-            )
-            return fn
-
-        return decorator
-
-
-class ToolExample(BaseModel):
-    """Example usage of a tool."""
-    name: str = Field(description="Name of the example")
-    description: str = Field(description="Description of what the example demonstrates")
-    code: str = Field(description="Example code showing how to use the tool")
-
-
-class EnhancedTool(FastMCPTool):
-    """Extended tool registration with rich metadata."""
-    examples: List[ToolExample] = Field(default_factory=list)
-    category: Optional[str] = Field(None)
-    input_schema: Optional[Dict] = Field(None)
-    output_schema: Optional[Dict] = Field(None)
-    output_model: Any = Field(None)
-
-    @staticmethod
-    def _get_schema(type_: Any) -> Optional[Dict[str, Any]]:
-        """Get JSON schema for a type using TypeAdapter."""
-        try:
-            # If it's already a dict schema, return it
-            if isinstance(type_, dict):
-                return type_
-                
-            # Handle simple types directly
-            if type_ in (str, int, float, bool):
-                return {"type": {
-                    str: "string",
-                    int: "integer",
-                    float: "number",
-                    bool: "boolean"
-                }[type_]}
-                
-            # Try using TypeAdapter for complex types
-            adapter = TypeAdapter(type_)
-            return adapter.json_schema()
-        except Exception as e:
-            logger.debug(f"Error getting schema for {type_}: {e}")
-            return None
-
-    @classmethod
-    def from_function(
-        cls,
-        fn: Callable,
-        name: Optional[str] = None,
-        description: Optional[str] = None,
-        examples: Optional[List[Dict]] = None,
-        category: Optional[str] = None,
-        output_model: Any = None,
-    ) -> "EnhancedTool":
-        """Create an enhanced tool from a function."""
-        # First create the base tool
-        base_tool = super().from_function(fn, name=name, description=description)
-
-        # Get output schema
-        output_schema = None
-        if output_model:
-            logger.debug(f"Getting schema for output_model {output_model}")
-            output_schema = cls._get_schema(output_model)
-        else:
-            # Try return type annotation
-            sig = inspect.signature(fn)
-            return_type = sig.return_annotation
-            if return_type != inspect.Signature.empty:
-                logger.debug(f"Getting schema from return type {return_type}")
-                output_schema = cls._get_schema(return_type)
-
-        # Provide basic schema if we couldn't get a proper one
-        if output_schema is None:
-            output_schema = {"type": "object"}
-
-        # Convert examples to ToolExample models
-        tool_examples = [ToolExample(**ex) for ex in (examples or [])]
-
-        return cls(
-            fn=fn,
-            name=base_tool.name,
-            description=base_tool.description,
-            parameters=base_tool.parameters,
-            fn_metadata=base_tool.fn_metadata,
-            is_async=base_tool.is_async,
-            context_kwarg=base_tool.context_kwarg,
-            examples=tool_examples,
-            category=category,
-            input_schema=base_tool.parameters,
-            output_schema=output_schema,
-            output_model=output_model,
-        )
-
-    def get_schema(self) -> Dict:
-        """Get complete tool schema including examples and metadata."""
-        return {
-            "name": self.name,
-            "description": self.description,
-            "category": self.category,
-            "input_schema": self.input_schema,
-            "output_schema": self.output_schema,
-            "examples": [ex.model_dump() for ex in self.examples],
-        }
-
-
-class EnhancedToolManager(FastMCPToolManager):
-    """Tool manager with enhanced metadata support."""
-
-    def add_tool(
-        self,
-        fn: Callable,
-        name: Optional[str] = None,
-        description: Optional[str] = None,
-        examples: Optional[List[Dict]] = None,
-        category: Optional[str] = None,
-        output_model: Any = None,
-    ) -> EnhancedTool:
-        """Add a tool with enhanced metadata."""
-        tool = EnhancedTool.from_function(
-            fn, 
-            name=name, 
-            description=description, 
-            examples=examples, 
-            category=category,
-            output_model=output_model
-        )
-        self._tools[tool.name] = tool
-        return tool
-
-    def get_schema_catalog(self) -> Dict:
-        """Get complete schema catalog for all tools."""
-        catalog = {"tools": {}, "categories": {}}
-
-        for tool in self._tools.values():
-            if isinstance(tool, EnhancedTool):
-                catalog["tools"][tool.name] = tool.get_schema()
-
-                if tool.category:
-                    if tool.category not in catalog["categories"]:
-                        catalog["categories"][tool.category] = {"name": tool.category, "tools": []}
-                    catalog["categories"][tool.category]["tools"].append(tool.name)
-
-        return catalog
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.utilities.logging import configure_logging
 
+configure_logging(level="INFO")
 
 # Create the shared server instance
-mcp = BasicMemoryServer("Basic Memory")
+mcp = FastMCP("Basic Memory")

src/basic_memory/mcp/tools/__init__.py

@@ -8,9 +8,8 @@ all tools with the MCP server.
 # Import tools to register them with MCP
 from basic_memory.mcp.tools import knowledge  # noqa: F401
 from basic_memory.mcp.tools import search     # noqa: F401
-from basic_memory.mcp.tools import discovery  # noqa: F401
 from basic_memory.mcp.tools import activity   # noqa: F401
-from basic_memory.mcp.tools.discussion import get_discussion_context
+from basic_memory.mcp.tools.discussion import build_context
 
 # Export the tools
 from basic_memory.mcp.tools.knowledge import (
@@ -20,17 +19,10 @@ from basic_memory.mcp.tools.knowledge import (
     delete_entities,
     delete_observations,
     delete_relations,
-    get_entity,
+    get_entity, 
+    get_entities,
 )
 
-from basic_memory.mcp.tools.search import (
-    search,
-    open_nodes,
-)
-
-from basic_memory.mcp.tools.discovery import (
-    get_observation_categories,
-)
 
 from basic_memory.mcp.tools.activity import (
     get_recent_activity,
@@ -45,18 +37,15 @@ __all__ = [
     "delete_entities",
     "delete_observations",
     "delete_relations",
+    "get_entity",
+    "get_entities",
 
     # Search tools
     "search",
-    "get_entity",
-    "open_nodes",
-
-    # Discovery tools
-    "get_observation_categories",
     
     # Activity tools
     "get_recent_activity",
     
     # memory tools
-    "get_discussion_context"
+    "build_context"
 ]

src/basic_memory/mcp/tools/activity.py

@@ -3,6 +3,7 @@
 from typing import List, Optional
 
 from loguru import logger
+from mcp.server.fastmcp import Context
 
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
@@ -10,114 +11,10 @@ from basic_memory.schemas.activity import ActivityType, RecentActivity
 
 
 @mcp.tool(
-    category="activity",
     description="Track recent changes to documents, entities, and relations",
-    examples=[
-        {
-            "name": "Activity Summary",
-            "description": "Get a high-level overview of changes",
-            "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": "Document Changes",
-            "description": "Track document evolution over time",
-            "code": """
-# Get hourly document changes with context
-docs = await get_recent_activity(
-    timeframe="1h",
-    activity_types=[ActivityType.DOCUMENT]
-)
-
-# Show document evolution chronologically
-for change in sorted(docs.changes, key=lambda x: x.timestamp):
-    print(f"{change.timestamp}: {change.permalink}")
-    print(f"  {change.change_type}: {change.summary}")
-""",
-        },
-        {
-            "name": "Knowledge Evolution",
-            "description": "Analyze how knowledge structure changes over time",
-            "code": """
-# Get weekly activity for change pattern analysis
-weekly = await get_recent_activity(timeframe="1w")
-
-# Group changes by type for pattern analysis
-from collections import defaultdict
-changes_by_type = defaultdict(list)
-for change in weekly.changes:
-    changes_by_type[change.activity_type].append(change)
-
-# Analyze change distribution
-for type_, changes in changes_by_type.items():
-    print(f"{type_}: {len(changes)} changes")
-    
-# Find most modified entities
-entity_changes = defaultdict(int)
-for change in weekly.changes:
-    if change.activity_type == "entity":
-        entity_changes[change.permalink] += 1
-
-print("\\nMost active entities:")
-for permalink, count in sorted(
-    entity_changes.items(), 
-    key=lambda x: x[1], 
-    reverse=True
-)[:5]:
-    print(f"- {permalink}: {count} changes")
-""",
-        },
-        {
-            "name": "Context Building",
-            "description": "Use activity history to build rich context",
-            "code": """
-# Get recent activity across all types
-activity = await get_recent_activity(timeframe="1d")
-
-# Extract changed entities for deeper analysis
-entity_ids = [
-    change.permalink for change in activity.changes
-    if change.activity_type == "entity"
-]
-
-# Load full entity details
-if entity_ids:
-    entities = await open_nodes(
-        request=OpenNodesRequest(permalinks=entity_ids)
-    )
-    
-    # Analyze recent development focus
-    tech_changes = defaultdict(list)
-    for entity in entities.entities:
-        tech_obs = [o for o in entity.observations 
-                   if o.category == "tech"]
-        if tech_obs:
-            tech_changes[entity.name] = tech_obs
-
-    print("Recent technical changes:")
-    for name, observations in tech_changes.items():
-        print(f"\\n{name}:")
-        for obs in observations:
-            print(f"- {obs.content}")
-""",
-        },
-    ],
-    output_model=RecentActivity,
 )
 async def get_recent_activity(
+    context: Context,
     timeframe: str = "1d",
     activity_types: Optional[List[ActivityType]] = None,
 ) -> RecentActivity:
@@ -126,11 +23,12 @@ async def get_recent_activity(
     Args:
         timeframe: Time window to analyze ("1h", "1d", "1w")
         activity_types: Optional list of types to filter by
+        context: MCP context 
 
     Returns:
         RecentActivity object with changes and summary statistics
     """
-    logger.debug(f"Getting recent activity (timeframe={timeframe}, types={activity_types})")
+    context.info(f"Getting recent activity (timeframe={timeframe}, types={activity_types})")
 
     # Build params
     params = {

src/basic_memory/mcp/tools/discovery.py

@@ -1,93 +0,0 @@
-"""Tools for discovering and analyzing knowledge graph structure."""
-
-from typing import List, Optional
-
-from loguru import logger
-
-from basic_memory.mcp.server import mcp
-from basic_memory.schemas import EntityTypeList, ObservationCategoryList, TypedEntityList
-from basic_memory.mcp.async_client import client
-
-
-
-@mcp.tool(
-    category="discovery",
-    description="List all unique observation categories used in the knowledge base",
-    examples=[
-        {
-            "name": "Category Usage",
-            "description": "Analyze how categories are used across entity types",
-            "code": """
-# Get all categories
-categories = await get_observation_categories()
-
-# Get usage patterns by entity type
-types = await get_entity_types()
-
-usage_patterns = defaultdict(lambda: defaultdict(int))
-for entity_type in types["types"]:
-    # Get entities of this type
-    entities = await list_by_type(entity_type)
-    
-    # Count category usage
-    for entity in entities.entities:
-        for obs in entity.observations:
-            usage_patterns[entity_type][obs.category] += 1
-
-# Show category usage patterns
-print("Category Usage Patterns:")
-for entity_type, patterns in usage_patterns.items():
-    if patterns:
-        print(f"\\n{entity_type}:")
-        for category, count in patterns.items():
-            print(f"- {category}: {count} observations")"""
-        },
-        {
-            "name": "Knowledge Organization",
-            "description": "Analyze knowledge organization patterns",
-            "code": """
-# Get all categories and entities
-categories = await get_observation_categories()
-results = await search_nodes(
-    request=SearchNodesRequest(
-        query="implementation architecture",
-        category=None  # Search all categories
-    )
-)
-
-# Analyze knowledge structure
-category_content = defaultdict(list)
-for entity in results.matches:
-    for obs in entity.observations:
-        category_content[obs.category].append({
-            "entity": entity.name,
-            "content": obs.content,
-            "context": obs.context
-        })
-
-# Show how knowledge is organized
-print("Knowledge Organization Analysis:")
-for category in categories:
-    content = category_content.get(category, [])
-    if content:
-        print(f"\\n{category.upper()} ({len(content)} items):")
-        # Show example content
-        examples = content[:3]
-        for ex in examples:
-            context = f" ({ex['context']})" if ex['context'] else ""
-            print(f"- {ex['entity']}: {ex['content']}{context}")"""
-        }
-    ],
-    output_model=List[str]
-)
-async def get_observation_categories() -> List[str]:
-    """List all unique observation categories in use.
-    
-    Returns:
-        List of unique category names used for observations
-    """
-    logger.debug("Getting all observation categories")
-    url = "/discovery/observation-categories"
-    response = await client.get(url)
-    return ObservationCategoryList.model_validate(response.json())
-

src/basic_memory/mcp/tools/discussion.py

@@ -2,80 +2,18 @@
 
 from typing import Optional
 
+from loguru import logger
+
 from basic_memory.mcp.async_client import client
 from basic_memory.mcp.server import mcp
 from basic_memory.schemas.memory import GraphContext, MemoryUrl
 
 
 @mcp.tool(
-    category="discussion",
-    description="Get discussion context from a memory:// URI to continue conversations naturally.",
-    examples=[
-        {
-            "name": "Continue Previous Discussion",
-            "description": "Load context to continue a technical discussion",
-            "code": """
-# Get context for previous discussion about search
-context = await get_discussion_context(
-    url="memory://specs/search-refactor",
-    depth=2, 
-    timeframe="7d"
+    name="Build Context",
+    description="Build context from a memory:// URI to continue conversations naturally.",
 )
-
-# Access the context components
-primary = context.primary_entities    # Main discussion topic
-related = context.related_entities    # Related concepts
-meta = context.metadata              # Discussion metadata
-
-# Analyze primary discussion topics
-print("\\nPrimary Discussion Topic:")
-for entity in primary:
-    print(f"- {entity.title}")
-    print(f"  Type: {entity.type}")
-    if "status" in entity.metadata:
-        print(f"  Status: {entity.metadata['status']}")
-
-# Analyze related content
-print("\\nRelated Topics:")
-for item in related:
-    print(f"- {item.title}")
-    if item.relation_type:
-        print(f"  Relation: {item.relation_type}")
-"""
-        },
-        {
-            "name": "Pattern Matching and Time Filtering",
-            "description": "Search for matching discussions within a timeframe",
-            "code": """
-# Find recent discussions matching a pattern
-context = await get_discussion_context(
-    url="memory://design/*",     # Match all design documents
-    depth=1,                     # Direct relations only
-    timeframe="3d"              # Last 3 days only
-)
-
-print(f"Found {context.metadata['matched_entities']} matching discussions")
-print(f"Total related items: {context.metadata['total_entities']}")
-
-# Group by document type
-by_type = {}
-for entity in context.primary_entities:
-    doc_type = entity.type
-    if doc_type not in by_type:
-        by_type[doc_type] = []
-    by_type[doc_type].append(entity)
-
-# Show summary by type
-for doc_type, entities in by_type.items():
-    print(f"\\n{doc_type.title()}:")
-    for entity in entities:
-        print(f"- {entity.title}")
-        print(f"  Added: {entity.metadata.get('created_at', 'Unknown')}")
-"""
-        }
-    ],
-)
-async def get_discussion_context(
+async def build_context(
     url: MemoryUrl,
     depth: Optional[int] = 2,
     timeframe: Optional[str] = "7d",
@@ -87,6 +25,7 @@ async def get_discussion_context(
     a rich context graph of related information.
 
     Args:
+        ctx: MCP context
         url: memory:// URI pointing to discussion content (e.g. memory://specs/search)
         depth: How many relation hops to traverse (default: 2)
         timeframe: How far back to look, e.g. "7d", "24h" (default: "7d")
@@ -97,6 +36,7 @@ async def get_discussion_context(
             - related_entities: Connected content via relations
             - metadata: Context building info
     """
+    logger.info(f"Building context from {url}")
     # Map directly to the memory endpoint
     memory_url = MemoryUrl.validate(url)
     response = await client.get(

src/basic_memory/mcp/tools/help.py

@@ -1,151 +0,0 @@
-"""Help and schema introspection tools."""
-
-from typing import Dict, List, Optional
-
-from pydantic import BaseModel, Field
-
-from basic_memory.mcp.server import mcp
-
-
-class ToolExample(BaseModel):
-    """Example usage of a tool."""
-    name: str = Field(description="Name of the example")
-    description: str = Field(description="Description of what the example demonstrates")
-    code: str = Field(description="Example code showing how to use the tool")
-
-
-class ToolSchema(BaseModel):
-    """Schema information for a single tool."""
-    name: str = Field(description="Name of the tool")
-    description: str = Field(description="Description of what the tool does")
-    category: Optional[str] = Field(None, description="Optional category for organization")
-    input_schema: Dict = Field(description="Schema for tool inputs")
-    output_schema: Dict = Field(description="Schema for tool outputs")
-    examples: List[ToolExample] = Field(
-        default_factory=list,
-        description="Example usages of the tool"
-    )
-
-
-class CategoryInfo(BaseModel):
-    """Information about a tool category."""
-    name: str = Field(description="Category name")
-    tools: List[str] = Field(description="Tools in this category")
-
-
-class SchemaCatalog(BaseModel):
-    """Complete schema catalog for all tools."""
-    tools: Dict[str, ToolSchema] = Field(
-        description="Map of tool names to their schemas"
-    )
-    categories: Dict[str, CategoryInfo] = Field(
-        default_factory=dict,
-        description="Tool categories and their tools"
-    )
-    referenced_models: Dict[str, Dict] = Field(
-        default_factory=dict,
-        description="Shared type definitions used by tools",
-        alias="referencedModels"
-    )
-
-    class Config:
-        """Pydantic config."""
-        json_schema_extra = {
-            "description": "Tool schema catalog showing available tools and their capabilities"
-        }
-
-
-@mcp.tool(
-    category="system",
-    description="Get schema information about available tools and their capabilities",
-    examples=[
-        {
-            "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": "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_model=SchemaCatalog
-)
-async def get_schema(
-    tool_name: Optional[str] = None, 
-    include_examples: bool = True, 
-    include_referenced: bool = True
-) -> Dict:
-    """Get schema information about available tools."""
-    # Our tool manager has the enhanced schema support
-    catalog = mcp._tool_manager.get_schema_catalog()
-
-    # Filter if specific tool requested
-    if tool_name:
-        if tool_name not in catalog["tools"]:
-            raise ValueError(f"Unknown tool: {tool_name}")
-
-        tool_schema = catalog["tools"][tool_name]
-
-        if not include_examples:
-            tool_schema.pop("examples", None)
-
-        if include_referenced:
-            response = {
-                "tools": {tool_name: tool_schema},
-                "referencedModels": tool_schema.get("referencedModels", {})
-            }
-        else:
-            response = {"tools": {tool_name: tool_schema}}
-
-        return SchemaCatalog.model_validate(response).model_dump()
-
-    # Return full catalog with requested inclusions
-    result = catalog
-
-    if not include_examples:
-        for tool in result["tools"].values():
-            tool.pop("examples", None)
-
-    return SchemaCatalog.model_validate(result).model_dump()

src/basic_memory/mcp/tools/knowledge.py

@@ -1,15 +1,16 @@
 """Knowledge graph management tools for Basic Memory MCP server."""
 
-from typing import Dict
-
 import httpx
+from loguru import logger
 
 from basic_memory.mcp.server import mcp
+from basic_memory.mcp.tools.utils import call_get
 from basic_memory.schemas.base import PathId
 from basic_memory.schemas.request import (
     CreateEntityRequest,
     CreateRelationsRequest,
     AddObservationsRequest,
+    GetEntitiesRequest,
 )
 from basic_memory.schemas.delete import (
     DeleteEntitiesRequest,
@@ -22,141 +23,39 @@ from basic_memory.services.exceptions import EntityNotFoundError
 
 
 @mcp.tool(
-    category="knowledge",
     description="Create new entities in the knowledge graph with names, types, and observations",
-    examples=[
-        {
-            "name": "Create Component",
-            "description": "Create a new technical component",
-            "code": """
-# Create search service component
-await create_entities({
-    "entities": [{
-        "name": "SearchService",
-        "entity_type": "component",
-        "description": "Full-text search capability",
-        "observations": [
-            "Implements FTS5 for better performance",
-            "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"
-        ]
-    }]
-})
-""",
-        },
-    ],
-    output_model=EntityListResponse,
 )
 async def create_entities(request: CreateEntityRequest) -> EntityListResponse:
     """Create new entities in the knowledge graph."""
+    logger.info(f"Creating {len(request.entities)} entities")
     url = "/knowledge/entities"
     response = await client.post(url, json=request.model_dump())
     return EntityListResponse.model_validate(response.json())
 
 
 @mcp.tool(
-    category="knowledge",
     description="Create typed relationships between existing entities",
-    examples=[
-        {
-            "name": "Add Dependency",
-            "description": "Create dependency relationship between components",
-            "code": """
-# Document component dependency
-await create_relations({
-    "relations": [{
-        "from_id": "component/search_service",
-        "to_id": "component/storage_service",
-        "relation_type": "depends_on",
-        "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"
-    }]
-})
-""",
-        },
-    ],
-    output_model=EntityListResponse,
 )
 async def create_relations(request: CreateRelationsRequest) -> EntityListResponse:
     """Create relations between existing entities."""
+    logger.info(f"Creating {len(request.relations)} relations")
     url = "/knowledge/relations"
     response = await client.post(url, json=request.model_dump())
     return EntityListResponse.model_validate(response.json())
 
 
 @mcp.tool(
-    category="knowledge",
     description="Get complete information about a specific entity including observations and relations",
-    examples=[
-        {
-            "name": "View Component Details",
-            "description": "Get complete component information",
-            "code": """
-# 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}")
-""",
-        }
-    ],
-    output_model=EntityResponse,
 )
-async def get_entity(permalink: PathId, content: bool = False) -> EntityResponse:
-    """Get a specific entity by its permalink.
+async def get_entity(permalink: PathId) -> EntityResponse:
+    """Get a specific entity info by its permalink.
 
     Args:
         permalink: Path identifier for the entity
-        content: If True, includes the full markdown content of the entity
     """
     try:
         url = f"/knowledge/entities/{permalink}"
-        params = {"content": "true"} if content else {}
-        response = await client.get(url, params=params)
+        response = await client.get(url)
         if response.status_code == 404:
             raise EntityNotFoundError(f"Entity not found: {permalink}")
         response.raise_for_status()
@@ -168,40 +67,30 @@ async def get_entity(permalink: PathId, content: bool = False) -> EntityResponse
 
 
 @mcp.tool(
-    description="Add categorized observations to an existing entity",
-    examples=[
-        {
-            "name": "Add Implementation Notes",
-            "description": "Document technical implementation details",
-            "code": """
-# Add technical observations
-await add_observations(
-    request=AddObservationsRequest(
-        permalink="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"
-            )
-        ]
-    )
+    description="Load multiple entities by their permalinks in a single request",
 )
-""",
-        }
-    ],
-    output_model=EntityResponse,
+async def get_entities(request: GetEntitiesRequest) -> EntityListResponse:
+    """Load multiple entities by their permalinks.
+
+    Args:
+        request: OpenNodesRequest containing list of permalinks to load
+
+    Returns:
+        EntityListResponse containing complete details for each requested entity
+    """
+    url = "/knowledge/entities"
+    response = await call_get(
+        client, url, params=[("permalink", permalink) for permalink in request.permalinks]
+    )
+    return EntityListResponse.model_validate(response.json())
+
+
+@mcp.tool(
+    description="Add categorized observations to an existing entity",
 )
 async def add_observations(request: AddObservationsRequest) -> EntityResponse:
     """Add observations to an existing entity."""
+    logger.info(f"Adding {len(request.observations)} observations to {request.permalink}")
     url = "/knowledge/observations"
     response = await client.post(url, json=request.model_dump())
     return EntityResponse.model_validate(response.json())
@@ -209,25 +98,6 @@ async def add_observations(request: AddObservationsRequest) -> EntityResponse:
 
 @mcp.tool(
     description="Delete specific observations from an entity while preserving other content",
-    examples=[
-        {
-            "name": "Remove Obsolete Notes",
-            "description": "Delete outdated observations",
-            "code": """
-# Remove old implementation notes
-await delete_observations(
-    request=DeleteObservationsRequest(
-        permalink="component/indexer",
-        observations=[
-            "Using old indexing algorithm",
-            "Temporary workaround for issue #123"
-        ]
-    )
-)
-""",
-        }
-    ],
-    output_model=EntityResponse,
 )
 async def delete_observations(request: DeleteObservationsRequest) -> EntityResponse:
     """Delete specific observations from an entity."""
@@ -238,25 +108,6 @@ async def delete_observations(request: DeleteObservationsRequest) -> EntityRespo
 
 @mcp.tool(
     description="Delete relationships between entities while preserving the entities themselves",
-    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"
-        }]
-    )
-)
-""",
-        }
-    ],
-    output_model=EntityListResponse,
 )
 async def delete_relations(request: DeleteRelationsRequest) -> EntityListResponse:
     """Delete relations between entities."""
@@ -267,24 +118,6 @@ async def delete_relations(request: DeleteRelationsRequest) -> EntityListRespons
 
 @mcp.tool(
     description="Permanently delete entities and all related content (observations and relations)",
-    examples=[
-        {
-            "name": "Remove Old Components",
-            "description": "Delete obsolete components",
-            "code": """
-# Remove deprecated components
-await delete_entities(
-    request=DeleteEntitiesRequest(
-        permalinks=[
-            "component/old_service",
-            "test/obsolete_test"
-        ]
-    )
-)
-""",
-        }
-    ],
-    output_model=Dict[str, bool],
 )
 async def delete_entities(request: DeleteEntitiesRequest) -> DeleteEntitiesResponse:
     """Delete entities from the knowledge graph."""

src/basic_memory/mcp/tools/search.py

@@ -1,161 +1,17 @@
 """Search tools for Basic Memory MCP server."""
+from mcp.server.fastmcp import Context
 
 from basic_memory.mcp.server import mcp
 from basic_memory.schemas.search import SearchQuery, SearchResponse
-from basic_memory.schemas.request import OpenNodesRequest
+from basic_memory.schemas.request import GetEntitiesRequest
 from basic_memory.schemas.response import EntityListResponse
 from basic_memory.mcp.async_client import client
 
 
 @mcp.tool(
-    category="search",
     description="Search across all content in basic-memory, including documents and entities",
-    examples=[
-        {
-            "name": "Basic Full-text Search with Analysis",
-            "description": "Search and analyze results by metadata categories",
-            "code": """
-# Full text search
-results = await search(
-    query=SearchQuery(
-        text="implementation"  # Full text query
-    )
 )
-
-# Group by status and type
-by_status = defaultdict(list)
-by_type = defaultdict(list)
-
-for result in results.results:
-    meta = result.metadata
-    path = result.permalink
-    
-    # Group by status if available
-    if "status" in meta:
-        by_status[meta["status"]].append(path)
-        
-    # Always group by type
-    by_type[result.type].append(path)
-
-print("\\nBy Status:")
-for status, paths in by_status.items():
-    print(f"\\n{status.title()}:")
-    for path in paths:
-        print(f"- {path}")
-
-print("\\nBy Type:")
-for type_, paths in by_type.items():
-    print(f"\\n{type_.title()}:")
-    for path in paths:
-        print(f"- {path}")
-""",
-        },
-        {
-            "name": "Recent Changes in Entity Types",
-            "description": "Find recent changes in specific entity types",
-            "code": """
-from datetime import datetime, timezone, timedelta
-
-# Set search parameters
-cutoff = datetime.now(timezone.utc) - timedelta(days=7)
-entity_types = ["component", "specification"]
-
-# Search for recent changes
-results = await search(
-    query=SearchQuery(
-        text="*",  # Match all
-        entity_types=entity_types,
-        after_date=cutoff.isoformat()
-    )
-)
-
-# Sort by update time
-sorted_results = sorted(
-    results.results,
-    key=lambda x: x.metadata.get("updated_at", ""),
-    reverse=True
-)
-
-print("Recent Changes:")
-for result in sorted_results:
-    print(f"\\n{result.permalink}")
-    print(f"Type: {result.type}")
-    print(f"Score: {result.score:.2f}")
-    if "updated_at" in result.metadata:
-        print(f"Updated: {result.metadata['updated_at']}")
-""",
-        },
-        {
-            "name": "Technical Documentation Search",
-            "description": "Search technical documentation with smart filtering",
-            "code": """
-# Search technical documentation
-results = await search(
-    query=SearchQuery(
-        text="database implementation",
-        types=["document"]  # Only documents
-    )
-)
-
-# Filter and process results
-docs = []
-for result in results.results:
-    meta = result.metadata
-    
-    # Include if it's a technical document
-    if (meta.get("category") in ["specification", "technical"] or
-        any(tag in meta.get("tags", []) for tag in ["technical", "spec", "documentation"])):
-        docs.append(result)
-
-# Sort by relevance score
-docs.sort(key=lambda x: x.score)
-
-print("Technical Documentation:")
-for doc in docs:
-    print(f"\\n{doc.permalink}")
-    if "title" in doc.metadata:
-        print(f"Title: {doc.metadata['title']}")
-    print(f"Score: {doc.score:.2f}")
-    if "tags" in doc.metadata:
-        print(f"Tags: {', '.join(doc.metadata['tags'])}")
-""",
-        },
-        {
-            "name": "Related Content Search",
-            "description": "Find content related to a specific entity",
-            "code": """
-# First get the entity to extract key terms
-entity = await get_entity(permalink="component/memory_service")
-
-if entity:
-    # Build search terms from entity info
-    search_terms = [
-        entity.get("name", ""),
-        *entity.get("tags", []),
-        entity.get("entity_type", "")
-    ]
-    
-    # Search using combined terms
-    results = await search(
-        query=SearchQuery(
-            text=" ".join(filter(None, search_terms))
-        )
-    )
-    
-    # Filter out the original entity and sort by relevance
-    related = [r for r in results.results if r.permalink != entity["permalink"]]
-    related.sort(key=lambda x: x.score)
-
-    print(f"Content Related to {entity['name']}:")
-    for result in related[:5]:  # Top 5 most relevant
-        print(f"\\n{result.permalink}")
-        print(f"Type: {result.type}")
-        print(f"Score: {result.score:.2f}")
-""",
-        },
-    ],
-)
-async def search(query: SearchQuery) -> SearchResponse:
+async def search(ctx: Context, query: SearchQuery) -> SearchResponse:
     """Search across all content in basic-memory.
 
     Args:
@@ -168,64 +24,6 @@ async def search(query: SearchQuery) -> SearchResponse:
     Returns:
         SearchResponse with search results and metadata
     """
+    ctx.info(f"Searching for {query.text}")
     response = await client.post("/search/", json=query.model_dump())
     return SearchResponse.model_validate(response.json())
-
-
-@mcp.tool(
-    category="search",
-    description="Load multiple entities by their permalinks in a single request",
-    examples=[
-        {
-            "name": "Load and Analyze Entity Context",
-            "description": "Load full entity details and analyze relationships",
-            "code": """
-# First search for related entities
-results = await search(
-    query=SearchQuery(
-        text="knowledge graph",
-        types=["entity"],
-        entity_types=["component", "concept"]
-    )
-)
-
-if results.results:
-    # Load full context for found entities
-    permalinks = [r.permalink for r in results.results]
-    context = await open_nodes(
-        request=OpenNodesRequest(permalinks=permalinks)
-    )
-    
-    # Analyze relationships
-    relationship_map = defaultdict(list)
-    for entity in context.entities:
-        print(f"\\n{entity.name} ({entity.entity_type})")
-        
-        # Group by relationship type
-        for relation in entity.relations:
-            relationship_map[relation.relation_type].append(
-                (entity.name, relation.to_id)
-            )
-    
-    # Show relationship summary
-    print("\\nRelationship Summary:")
-    for rel_type, connections in relationship_map.items():
-        print(f"\\n{rel_type}:")
-        for source, target in connections:
-            print(f"- {source} -> {target}")
-""",
-        }
-    ],
-)
-async def open_nodes(request: OpenNodesRequest) -> EntityListResponse:
-    """Load multiple entities by their permalinks.
-
-    Args:
-        request: OpenNodesRequest containing list of permalinks to load
-
-    Returns:
-        EntityListResponse containing complete details for each requested entity
-    """
-    url = "/knowledge/nodes"
-    response = await client.post(url, json=request.model_dump())
-    return EntityListResponse.model_validate(response.json())

src/basic_memory/mcp/tools/utils.py

@@ -0,0 +1,123 @@
+import typing
+
+from httpx import Response, URL, AsyncClient, HTTPStatusError
+from httpx._client import UseClientDefault, USE_CLIENT_DEFAULT
+from httpx._types import (
+    RequestContent,
+    RequestData,
+    RequestFiles,
+    QueryParamTypes,
+    HeaderTypes,
+    CookieTypes,
+    AuthTypes,
+    TimeoutTypes,
+    RequestExtensions,
+)
+from loguru import logger
+from mcp.server.fastmcp.exceptions import ToolError
+
+
+async def call_get(
+    client: AsyncClient,
+    url: URL | str,
+    *,
+    params: QueryParamTypes | None = None,
+    headers: HeaderTypes | None = None,
+    cookies: CookieTypes | None = None,
+    auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
+    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
+    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    extensions: RequestExtensions | None = None,
+) -> Response:
+    try:
+        response = await client.get(
+            url,
+            params=params,
+            headers=headers,
+            cookies=cookies,
+            auth=auth,
+            follow_redirects=follow_redirects,
+            timeout=timeout,
+            extensions=extensions,
+        )
+        response.raise_for_status()
+        return response
+    except HTTPStatusError as e:
+        logger.error(f"Error calling GET {url}: {e}")
+        raise ToolError(f"Error calling tool: {e}") from e
+
+
+async def call_put(
+    client: AsyncClient,
+    url: URL | str,
+    *,
+    content: RequestContent | None = None,
+    data: RequestData | None = None,
+    files: RequestFiles | None = None,
+    json: typing.Any | None = None,
+    params: QueryParamTypes | None = None,
+    headers: HeaderTypes | None = None,
+    cookies: CookieTypes | None = None,
+    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
+    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    extensions: RequestExtensions | None = None,
+) -> Response:
+    try:
+        response = await client.put(
+            url,
+            content=content,
+            data=data,
+            files=files,
+            json=json,
+            params=params,
+            headers=headers,
+            cookies=cookies,
+            auth=auth,
+            follow_redirects=follow_redirects,
+            timeout=timeout,
+            extensions=extensions,
+        )
+        response.raise_for_status()
+        return response
+    except HTTPStatusError as e:
+        logger.error(f"Error calling PUT {url}: {e}")
+        raise ToolError(f"Error calling tool: {e}") from e
+
+
+async def call_post(
+    client: AsyncClient,
+    url: URL | str,
+    *,
+    content: RequestContent | None = None,
+    data: RequestData | None = None,
+    files: RequestFiles | None = None,
+    json: typing.Any | None = None,
+    params: QueryParamTypes | None = None,
+    headers: HeaderTypes | None = None,
+    cookies: CookieTypes | None = None,
+    auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
+    timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
+    extensions: RequestExtensions | None = None,
+) -> Response:
+    try:
+        response = await client.post(
+            url=url,
+            content=content,
+            data=data,
+            files=files,
+            json=json,
+            params=params,
+            headers=headers,
+            cookies=cookies,
+            auth=auth,
+            follow_redirects=follow_redirects,
+            timeout=timeout,
+            extensions=extensions,
+        )
+        response.raise_for_status()
+        return response
+    except HTTPStatusError as e:
+        logger.error(f"Error calling POST {url}: {e}")
+        raise ToolError(f"Error calling tool: {e}") from e

src/basic_memory/schemas/__init__.py

@@ -26,7 +26,7 @@ from basic_memory.schemas.request import (
     AddObservationsRequest,
     CreateEntityRequest,
     SearchNodesRequest,
-    OpenNodesRequest,
+    GetEntitiesRequest,
     CreateRelationsRequest, UpdateEntityRequest,
 )
 
@@ -59,7 +59,7 @@ __all__ = [
     "AddObservationsRequest",
     "CreateEntityRequest",
     "SearchNodesRequest",
-    "OpenNodesRequest",
+    "GetEntitiesRequest",
     "CreateRelationsRequest",
     "UpdateEntityRequest",
     # Responses

src/basic_memory/schemas/request.py

@@ -80,7 +80,7 @@ class SearchNodesRequest(BaseModel):
     category: Optional[ObservationCategory] = None
 
 
-class OpenNodesRequest(BaseModel):
+class GetEntitiesRequest(BaseModel):
     """Retrieve specific entities by their IDs.
 
     Used to load complete entity details including all observations

src/basic_memory/services/entity_service.py

@@ -188,9 +188,9 @@ class EntityService(BaseService[EntityModel]):
         logger.debug(f"Listing entities: type={entity_type} sort={sort_by}")
         return await self.repository.list_entities(entity_type=entity_type, sort_by=sort_by)
 
-    async def open_nodes(self, permalinks: List[str]) -> Sequence[EntityModel]:
+    async def get_entities_by_permalinks(self, permalinks: List[str]) -> Sequence[EntityModel]:
         """Get specific nodes and their relationships."""
-        logger.debug(f"Opening nodes permalinks: {permalinks}")
+        logger.debug(f"Getting entities permalinks: {permalinks}")
         return await self.repository.find_by_permalinks(permalinks)
 
     async def delete_entity_by_file_path(self, file_path):