mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 963370373f | |||
| d1ca37fcf4 |
@@ -8,7 +8,7 @@ from fastmcp import Context
|
||||
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.schemas.base import TimeFrame
|
||||
from basic_memory.schemas.memory import GraphContext, MemoryUrl
|
||||
from basic_memory.schemas.memory import GraphContext, MemoryUrl, parse_memory_url
|
||||
|
||||
|
||||
@mcp.tool(
|
||||
@@ -19,9 +19,10 @@ from basic_memory.schemas.memory import GraphContext, MemoryUrl
|
||||
Memory URL Format:
|
||||
- Use paths like "folder/note" or "memory://folder/note"
|
||||
- Pattern matching: "folder/*" matches all notes in folder
|
||||
- Query parameters: "memory://folder/note?project=myproject" to target a specific project
|
||||
- Valid characters: letters, numbers, hyphens, underscores, forward slashes
|
||||
- Avoid: double slashes (//), angle brackets (<>), quotes, pipes (|)
|
||||
- Examples: "specs/search", "projects/basic-memory", "notes/*"
|
||||
- Examples: "specs/search", "memory://specs/search?project=research", "notes/*"
|
||||
|
||||
Timeframes support natural language like:
|
||||
- "2 days ago", "last week", "today", "3 months ago"
|
||||
@@ -52,7 +53,9 @@ async def build_context(
|
||||
Args:
|
||||
project: Project name to build context from. Optional - server will resolve using hierarchy.
|
||||
If unknown, use list_memory_projects() to discover available projects.
|
||||
Can also be specified via ?project= query parameter on the URL.
|
||||
url: memory:// URI pointing to discussion content (e.g. memory://specs/search)
|
||||
Supports ?project= query parameter (e.g. memory://specs/search?project=research)
|
||||
depth: How many relation hops to traverse (1-3 recommended for performance)
|
||||
timeframe: How far back to look. Supports natural language like "2 days ago", "last week"
|
||||
page: Page number of results to return (default: 1)
|
||||
@@ -82,6 +85,11 @@ async def build_context(
|
||||
Raises:
|
||||
ToolError: If project doesn't exist or depth parameter is invalid
|
||||
"""
|
||||
# Extract ?project= from URL when no explicit project parameter was provided
|
||||
url, url_params = parse_memory_url(url)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
logger.info(f"Building context from {url} in project {project}")
|
||||
|
||||
# Convert string depth to integer if needed
|
||||
@@ -93,8 +101,6 @@ async def build_context(
|
||||
|
||||
raise ToolError(f"Invalid depth parameter: '{depth}' is not a valid integer")
|
||||
|
||||
# URL is already validated and normalized by MemoryUrl type annotation
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
# Resolve memory:// identifier with project-prefix awareness
|
||||
_, resolved_path, _ = await resolve_project_and_path(
|
||||
|
||||
@@ -7,6 +7,7 @@ from mcp.server.fastmcp.exceptions import ToolError
|
||||
|
||||
from basic_memory.mcp.project_context import get_project_client
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.schemas.memory import parse_memory_url
|
||||
|
||||
|
||||
def _format_delete_error_response(project: str, error_message: str, identifier: str) -> str:
|
||||
@@ -215,6 +216,12 @@ async def delete_note(
|
||||
with suggestions for finding the correct identifier, including search
|
||||
commands and alternative formats to try.
|
||||
"""
|
||||
# Extract ?project= from identifier when no explicit project parameter was provided
|
||||
if identifier.startswith("memory://"):
|
||||
identifier, url_params = parse_memory_url(identifier)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
logger.debug(
|
||||
f"Deleting {'directory' if is_directory else 'note'}: {identifier} in project: {active_project.name}"
|
||||
|
||||
@@ -7,6 +7,7 @@ from fastmcp import Context
|
||||
|
||||
from basic_memory.mcp.project_context import get_project_client, add_project_metadata
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.schemas.memory import parse_memory_url
|
||||
|
||||
|
||||
def _format_error_response(
|
||||
@@ -211,6 +212,12 @@ async def edit_note(
|
||||
search_notes() first to find the correct identifier. The tool provides detailed
|
||||
error messages with suggestions if operations fail.
|
||||
"""
|
||||
# Extract ?project= from identifier when no explicit project parameter was provided
|
||||
if identifier.startswith("memory://"):
|
||||
identifier, url_params = parse_memory_url(identifier)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
logger.info("MCP tool call", tool="edit_note", identifier=identifier, operation=operation)
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ from fastmcp import Context
|
||||
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.project_context import get_project_client
|
||||
from basic_memory.schemas.memory import parse_memory_url
|
||||
from basic_memory.utils import validate_project_path
|
||||
|
||||
|
||||
@@ -411,6 +412,12 @@ async def move_note(
|
||||
- Re-indexes the entity for search
|
||||
- Maintains all observations and relations
|
||||
"""
|
||||
# Extract ?project= from identifier when no explicit project parameter was provided
|
||||
if identifier.startswith("memory://"):
|
||||
identifier, url_params = parse_memory_url(identifier)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
logger.debug(
|
||||
f"Moving {'directory' if is_directory else 'note'}: {identifier} to {destination_path} in project: {active_project.name}"
|
||||
|
||||
@@ -18,7 +18,7 @@ from mcp.server.fastmcp.exceptions import ToolError
|
||||
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.tools.utils import call_get, resolve_entity_id
|
||||
from basic_memory.schemas.memory import memory_url_path
|
||||
from basic_memory.schemas.memory import memory_url_path, parse_memory_url
|
||||
from basic_memory.utils import validate_project_path
|
||||
|
||||
|
||||
@@ -199,6 +199,12 @@ async def read_content(
|
||||
HTTPError: If project doesn't exist or is inaccessible
|
||||
SecurityError: If path attempts path traversal
|
||||
"""
|
||||
# Extract ?project= from path when no explicit project parameter was provided
|
||||
if path.startswith("memory://"):
|
||||
path, url_params = parse_memory_url(path)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
logger.info("Reading file", path=path, project=project)
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
|
||||
@@ -10,7 +10,7 @@ from basic_memory.mcp.project_context import get_project_client, resolve_project
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.mcp.formatting import format_note_preview_ascii
|
||||
from basic_memory.mcp.tools.search import search_notes
|
||||
from basic_memory.schemas.memory import memory_url_path
|
||||
from basic_memory.schemas.memory import memory_url_path, parse_memory_url
|
||||
from basic_memory.utils import validate_project_path
|
||||
|
||||
|
||||
@@ -81,6 +81,12 @@ async def read_note(
|
||||
If the exact note isn't found, this tool provides helpful suggestions
|
||||
including related notes, search commands, and note creation templates.
|
||||
"""
|
||||
# Extract ?project= from identifier when no explicit project parameter was provided
|
||||
if identifier.startswith("memory://"):
|
||||
identifier, url_params = parse_memory_url(identifier)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
async with get_project_client(project, context) as (client, active_project):
|
||||
# Resolve identifier with project-prefix awareness for memory:// URLs
|
||||
_, entity_path, _ = await resolve_project_and_path(
|
||||
|
||||
@@ -9,6 +9,7 @@ from fastmcp import Context
|
||||
from basic_memory.mcp.project_context import get_project_client, resolve_project_and_path
|
||||
from basic_memory.mcp.formatting import format_search_results_ascii
|
||||
from basic_memory.mcp.server import mcp
|
||||
from basic_memory.schemas.memory import parse_memory_url
|
||||
from basic_memory.schemas.search import (
|
||||
SearchItemType,
|
||||
SearchQuery,
|
||||
@@ -395,6 +396,12 @@ async def search_notes(
|
||||
# Explicit project specification
|
||||
results = await search_notes("project planning", project="my-project")
|
||||
"""
|
||||
# Extract ?project= from query when it's a memory:// URL
|
||||
if query.startswith("memory://"):
|
||||
query, url_params = parse_memory_url(query)
|
||||
if project is None and "project" in url_params:
|
||||
project = url_params["project"]
|
||||
|
||||
# Avoid mutable-default-argument footguns. Treat None as "no filter".
|
||||
types = types or []
|
||||
entity_types = entity_types or []
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from datetime import datetime
|
||||
from typing import List, Optional, Annotated, Sequence, Literal, Union, Dict
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
from annotated_types import MinLen, MaxLen
|
||||
from pydantic import BaseModel, Field, BeforeValidator, TypeAdapter, field_serializer
|
||||
@@ -13,7 +14,7 @@ def validate_memory_url_path(path: str) -> bool:
|
||||
"""Validate that a memory URL path is well-formed.
|
||||
|
||||
Args:
|
||||
path: The path part of a memory URL (without memory:// prefix)
|
||||
path: The path part of a memory URL (without memory:// prefix or query string)
|
||||
|
||||
Returns:
|
||||
True if the path is valid, False otherwise
|
||||
@@ -38,22 +39,69 @@ def validate_memory_url_path(path: str) -> bool:
|
||||
if "//" in path:
|
||||
return False
|
||||
|
||||
# Check for invalid characters (excluding * which is used for pattern matching)
|
||||
invalid_chars = {"<", ">", '"', "|", "?"}
|
||||
# Check for invalid characters (excluding * for pattern matching and ? for query strings)
|
||||
invalid_chars = {"<", ">", '"', "|"}
|
||||
if any(char in path for char in invalid_chars):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _split_query_string(url: str) -> tuple[str, str]:
|
||||
"""Split a URL into path and query string portions.
|
||||
|
||||
Returns:
|
||||
Tuple of (path_portion, query_string) where query_string excludes the leading '?'.
|
||||
If no query string is present, query_string is empty.
|
||||
"""
|
||||
if "?" in url:
|
||||
path, query_string = url.split("?", 1)
|
||||
return path, query_string
|
||||
return url, ""
|
||||
|
||||
|
||||
def parse_memory_url(url: str) -> tuple[str, dict[str, str]]:
|
||||
"""Parse a memory:// URL into its path and query parameters.
|
||||
|
||||
Extracts query parameters (like ?project=myproject) from a memory URL,
|
||||
returning the clean URL and a dict of single-valued params.
|
||||
|
||||
Args:
|
||||
url: A memory URL, possibly with query parameters.
|
||||
|
||||
Returns:
|
||||
Tuple of (clean_memory_url, params_dict).
|
||||
The clean URL has query parameters stripped.
|
||||
Params are flattened to single values (last value wins for duplicates).
|
||||
|
||||
Examples:
|
||||
>>> parse_memory_url("memory://specs/search?project=research")
|
||||
('memory://specs/search', {'project': 'research'})
|
||||
>>> parse_memory_url("memory://specs/search")
|
||||
('memory://specs/search', {})
|
||||
>>> parse_memory_url("specs/search?project=foo")
|
||||
('specs/search', {'project': 'foo'})
|
||||
"""
|
||||
base, query_string = _split_query_string(url)
|
||||
if not query_string:
|
||||
return url, {}
|
||||
|
||||
# parse_qs returns lists; flatten to single values (last wins)
|
||||
parsed = parse_qs(query_string, keep_blank_values=False)
|
||||
params = {k: v[-1] for k, v in parsed.items()}
|
||||
return base, params
|
||||
|
||||
|
||||
def normalize_memory_url(url: str | None) -> str:
|
||||
"""Normalize a MemoryUrl string with validation.
|
||||
|
||||
Query parameters (e.g. ?project=foo) are preserved through normalization.
|
||||
|
||||
Args:
|
||||
url: A path like "specs/search" or "memory://specs/search"
|
||||
url: A path like "specs/search" or "memory://specs/search?project=foo"
|
||||
|
||||
Returns:
|
||||
Normalized URL starting with memory://
|
||||
Normalized URL starting with memory://, with any query params preserved
|
||||
|
||||
Raises:
|
||||
ValueError: If the URL path is malformed
|
||||
@@ -61,8 +109,8 @@ def normalize_memory_url(url: str | None) -> str:
|
||||
Examples:
|
||||
>>> normalize_memory_url("specs/search")
|
||||
'memory://specs/search'
|
||||
>>> normalize_memory_url("memory://specs/search")
|
||||
'memory://specs/search'
|
||||
>>> normalize_memory_url("memory://specs/search?project=foo")
|
||||
'memory://specs/search?project=foo'
|
||||
>>> normalize_memory_url("memory//test")
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
@@ -77,9 +125,12 @@ def normalize_memory_url(url: str | None) -> str:
|
||||
if not url:
|
||||
raise ValueError("Memory URL cannot be empty or whitespace")
|
||||
|
||||
clean_path = url.removeprefix("memory://")
|
||||
# Separate query string before validation — query params are not part of the path
|
||||
base, query_string = _split_query_string(url)
|
||||
|
||||
# Validate the extracted path
|
||||
clean_path = base.removeprefix("memory://")
|
||||
|
||||
# Validate the path portion only (query string is handled separately)
|
||||
if not validate_memory_url_path(clean_path):
|
||||
# Provide specific error messages for common issues
|
||||
if "://" in clean_path:
|
||||
@@ -89,7 +140,10 @@ def normalize_memory_url(url: str | None) -> str:
|
||||
else:
|
||||
raise ValueError(f"Invalid memory URL path: '{clean_path}' contains invalid characters")
|
||||
|
||||
return f"memory://{clean_path}"
|
||||
normalized = f"memory://{clean_path}"
|
||||
if query_string:
|
||||
normalized = f"{normalized}?{query_string}"
|
||||
return normalized
|
||||
|
||||
|
||||
MemoryUrl = Annotated[
|
||||
@@ -105,19 +159,20 @@ memory_url = TypeAdapter(MemoryUrl)
|
||||
|
||||
def memory_url_path(url: memory_url) -> str: # pyright: ignore
|
||||
"""
|
||||
Returns the uri for a url value by removing the prefix "memory://" from a given MemoryUrl.
|
||||
Returns the path for a url value by removing the "memory://" prefix and any query string.
|
||||
|
||||
This function processes a given MemoryUrl by removing the "memory://"
|
||||
prefix and returns the resulting string. If the provided url does not
|
||||
begin with "memory://", the function will simply return the input url
|
||||
unchanged.
|
||||
Strips both the memory:// scheme and any query parameters (?project=foo etc.)
|
||||
to return just the path portion used for entity lookup.
|
||||
|
||||
:param url: A MemoryUrl object representing the URL with a "memory://" prefix.
|
||||
:type url: MemoryUrl
|
||||
:return: A string representing the URL with the "memory://" prefix removed.
|
||||
:return: The path portion of the URL, without prefix or query string.
|
||||
:rtype: str
|
||||
"""
|
||||
return url.removeprefix("memory://")
|
||||
path = url.removeprefix("memory://")
|
||||
# Strip query string — callers need just the path for lookups
|
||||
path, _ = _split_query_string(path)
|
||||
return path
|
||||
|
||||
|
||||
class EntitySummary(BaseModel):
|
||||
|
||||
@@ -7,6 +7,8 @@ from basic_memory.schemas.memory import (
|
||||
normalize_memory_url,
|
||||
validate_memory_url_path,
|
||||
memory_url,
|
||||
memory_url_path,
|
||||
parse_memory_url,
|
||||
)
|
||||
|
||||
|
||||
@@ -83,7 +85,6 @@ class TestValidateMemoryUrlPath:
|
||||
"notes<with>brackets",
|
||||
'notes"with"quotes',
|
||||
"notes|with|pipes",
|
||||
"notes?with?questions",
|
||||
]
|
||||
|
||||
for path in invalid_paths:
|
||||
@@ -91,6 +92,10 @@ class TestValidateMemoryUrlPath:
|
||||
f"Path '{path}' should be invalid (invalid chars)"
|
||||
)
|
||||
|
||||
def test_question_mark_allowed_in_path(self):
|
||||
"""Test that ? is allowed since it serves as query string separator."""
|
||||
assert validate_memory_url_path("notes?project=foo")
|
||||
|
||||
|
||||
class TestNormalizeMemoryUrl:
|
||||
"""Test the normalize_memory_url function."""
|
||||
@@ -112,6 +117,26 @@ class TestNormalizeMemoryUrl:
|
||||
f"normalize_memory_url('{input_url}') should return '{expected}', got '{result}'"
|
||||
)
|
||||
|
||||
def test_query_params_preserved(self):
|
||||
"""Test that query parameters are preserved through normalization."""
|
||||
test_cases = [
|
||||
("specs/search?project=research", "memory://specs/search?project=research"),
|
||||
(
|
||||
"memory://specs/search?project=research",
|
||||
"memory://specs/search?project=research",
|
||||
),
|
||||
(
|
||||
"notes/meeting?project=work&depth=2",
|
||||
"memory://notes/meeting?project=work&depth=2",
|
||||
),
|
||||
]
|
||||
|
||||
for input_url, expected in test_cases:
|
||||
result = normalize_memory_url(input_url)
|
||||
assert result == expected, (
|
||||
f"normalize_memory_url('{input_url}') should return '{expected}', got '{result}'"
|
||||
)
|
||||
|
||||
def test_empty_url(self):
|
||||
"""Test that empty URLs raise ValueError."""
|
||||
with pytest.raises(ValueError, match="cannot be empty"):
|
||||
@@ -164,7 +189,6 @@ class TestNormalizeMemoryUrl:
|
||||
"notes<brackets>",
|
||||
'notes"quotes"',
|
||||
"notes|pipes|",
|
||||
"notes?questions?",
|
||||
]
|
||||
|
||||
for url in invalid_urls:
|
||||
@@ -183,6 +207,8 @@ class TestMemoryUrlPydanticValidation:
|
||||
"notes/meeting-2025",
|
||||
"projects/basic-memory/docs",
|
||||
"pattern/*",
|
||||
"specs/search?project=research",
|
||||
"memory://specs/search?project=research",
|
||||
]
|
||||
|
||||
for url in valid_urls:
|
||||
@@ -272,3 +298,63 @@ class TestMemoryUrlErrorMessages:
|
||||
error_msg = str(exc_info.value)
|
||||
assert "notes<brackets>" in error_msg
|
||||
assert "invalid characters" in error_msg
|
||||
|
||||
|
||||
class TestParseMemoryUrl:
|
||||
"""Test the parse_memory_url function."""
|
||||
|
||||
def test_url_without_params(self):
|
||||
"""Test parsing a URL with no query parameters."""
|
||||
url, params = parse_memory_url("memory://specs/search")
|
||||
assert url == "memory://specs/search"
|
||||
assert params == {}
|
||||
|
||||
def test_url_with_project_param(self):
|
||||
"""Test parsing a URL with ?project= query parameter."""
|
||||
url, params = parse_memory_url("memory://specs/search?project=research")
|
||||
assert url == "memory://specs/search"
|
||||
assert params == {"project": "research"}
|
||||
|
||||
def test_url_with_multiple_params(self):
|
||||
"""Test parsing a URL with multiple query parameters."""
|
||||
url, params = parse_memory_url("memory://specs/search?project=research&depth=2")
|
||||
assert url == "memory://specs/search"
|
||||
assert params == {"project": "research", "depth": "2"}
|
||||
|
||||
def test_bare_path_with_params(self):
|
||||
"""Test parsing a bare path (no memory:// prefix) with query parameters."""
|
||||
url, params = parse_memory_url("specs/search?project=foo")
|
||||
assert url == "specs/search"
|
||||
assert params == {"project": "foo"}
|
||||
|
||||
def test_duplicate_params_last_wins(self):
|
||||
"""Test that duplicate query parameters use last value."""
|
||||
url, params = parse_memory_url("memory://specs/search?project=a&project=b")
|
||||
assert url == "memory://specs/search"
|
||||
assert params == {"project": "b"}
|
||||
|
||||
def test_empty_param_values_excluded(self):
|
||||
"""Test that empty parameter values are excluded."""
|
||||
url, params = parse_memory_url("memory://specs/search?project=")
|
||||
assert url == "memory://specs/search"
|
||||
assert params == {}
|
||||
|
||||
|
||||
class TestMemoryUrlPathWithQueryParams:
|
||||
"""Test that memory_url_path strips query parameters."""
|
||||
|
||||
def test_path_without_params(self):
|
||||
"""Test extracting path from URL without query params."""
|
||||
assert memory_url_path("memory://specs/search") == "specs/search"
|
||||
|
||||
def test_path_strips_query_params(self):
|
||||
"""Test that query params are stripped when extracting the path."""
|
||||
assert memory_url_path("memory://specs/search?project=research") == "specs/search"
|
||||
|
||||
def test_path_strips_multiple_params(self):
|
||||
"""Test stripping multiple query parameters."""
|
||||
assert memory_url_path("memory://notes/meeting?project=work&depth=2") == "notes/meeting"
|
||||
|
||||
def test_no_prefix(self):
|
||||
"""Test path extraction when no memory:// prefix."""
|
||||
assert memory_url_path("specs/search?project=foo") == "specs/search"
|
||||
|
||||
Reference in New Issue
Block a user