diff --git a/scripts/coding/ai/mcp/__init__.py b/scripts/coding/ai/mcp/__init__.py new file mode 100644 index 00000000..ae48bc7b --- /dev/null +++ b/scripts/coding/ai/mcp/__init__.py @@ -0,0 +1,31 @@ +"""Helpers for configuring MCP servers and registries.""" + +from .memory import ( + MCPServerMemoryProfile, + MemoryBackendConfig, + MemoryEntry, + MemoryLayerConfig, + MemoryRetentionPolicy, + MemoryStore, + MemoryType, +) +from .registry import ( + MCPRegistry, + LocalMCPServer, + RemoteMCPServer, + build_default_registry, +) + +__all__ = [ + "MCPServerMemoryProfile", + "MCPRegistry", + "LocalMCPServer", + "MemoryBackendConfig", + "MemoryEntry", + "MemoryLayerConfig", + "MemoryRetentionPolicy", + "MemoryStore", + "MemoryType", + "RemoteMCPServer", + "build_default_registry", +] diff --git a/scripts/coding/ai/mcp/memory.py b/scripts/coding/ai/mcp/memory.py new file mode 100644 index 00000000..6ea8621f --- /dev/null +++ b/scripts/coding/ai/mcp/memory.py @@ -0,0 +1,186 @@ +"""Memory primitives so MCP servers match the agent experience.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from enum import Enum +from typing import Dict, List, Mapping, Optional, Tuple + + +class MemoryType(str, Enum): + """Memory taxonomy aligned with the agent documentation.""" + + WORKING = "working" + SHORT_TERM = "short_term" + LONG_TERM = "long_term" + PERSONA = "persona" + EPISODIC = "episodic" + ENTITY = "entity" + + +@dataclass(frozen=True) +class MemoryRetentionPolicy: + """Retention behavior for a memory lane.""" + + max_entries: Optional[int] = None + ttl_seconds: Optional[int] = None + + def is_expired(self, created_at: datetime, now: Optional[datetime] = None) -> bool: + if self.ttl_seconds is None: + return False + now = now or datetime.now(UTC) + return (now - created_at) > timedelta(seconds=self.ttl_seconds) + + def as_dict(self) -> Dict[str, Optional[int]]: + return {"max_entries": self.max_entries, "ttl_seconds": self.ttl_seconds} + + +@dataclass(frozen=True) +class MemoryBackendConfig: + """Where the memory lives (Mem0, Azure AI Search, etc.).""" + + provider: str + description: str + capabilities: Tuple[str, ...] = () + parameters: Mapping[str, str] = field(default_factory=dict) + + def as_dict(self) -> Dict[str, object]: + payload: Dict[str, object] = { + "provider": self.provider, + "description": self.description, + } + if self.capabilities: + payload["capabilities"] = list(self.capabilities) + if self.parameters: + payload["parameters"] = dict(self.parameters) + return payload + + +@dataclass(frozen=True) +class MemoryLayerConfig: + """Glue retention + backend information with a description.""" + + memory_type: MemoryType + retention: MemoryRetentionPolicy + storage: MemoryBackendConfig + purpose: str + tools: Tuple[str, ...] = () + + def as_dict(self) -> Dict[str, object]: + payload: Dict[str, object] = { + "memory_type": self.memory_type.value, + "retention": self.retention.as_dict(), + "storage": self.storage.as_dict(), + "purpose": self.purpose, + } + if self.tools: + payload["tools"] = list(self.tools) + return payload + + +@dataclass(frozen=True) +class MCPServerMemoryProfile: + """Full memory definition for a server.""" + + layers: Tuple[MemoryLayerConfig, ...] + structured_rag: Optional[MemoryBackendConfig] = None + + def as_dict(self) -> Dict[str, object]: + payload: Dict[str, object] = { + "layers": [layer.as_dict() for layer in self.layers], + } + if self.structured_rag: + payload["structured_rag"] = self.structured_rag.as_dict() + return payload + + +@dataclass +class MemoryEntry: + """Concrete memory instance stored in runtime.""" + + memory_type: MemoryType + content: str + metadata: Mapping[str, object] = field(default_factory=dict) + timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) + + def matches_keyword(self, keyword: str) -> bool: + needle = keyword.lower() + if needle in self.content.lower(): + return True + return any(needle in str(value).lower() for value in self.metadata.values()) + + +class MemoryStore: + """In-memory storage used for working + short term buffers.""" + + def __init__(self) -> None: + self._entries: List[MemoryEntry] = [] + + def add_entry(self, entry: MemoryEntry) -> None: + self._entries.append(entry) + + def get_entries(self, memory_type: Optional[MemoryType] = None) -> List[MemoryEntry]: + entries = [ + entry + for entry in self._entries + if memory_type is None or entry.memory_type == memory_type + ] + return sorted(entries, key=lambda entry: entry.timestamp) + + def search(self, keyword: str, memory_type: Optional[MemoryType] = None) -> List[MemoryEntry]: + return [ + entry + for entry in self.get_entries(memory_type) + if entry.matches_keyword(keyword) + ] + + def prune( + self, + policy: MemoryRetentionPolicy, + memory_type: Optional[MemoryType] = None, + now: Optional[datetime] = None, + ) -> None: + now = now or datetime.now(UTC) + survivors: List[MemoryEntry] = [] + for entry in self._entries: + if memory_type is not None and entry.memory_type != memory_type: + survivors.append(entry) + continue + if policy.is_expired(entry.timestamp, now): + continue + survivors.append(entry) + self._entries = survivors + + if policy.max_entries is None: + return + + if memory_type is not None: + self._enforce_max_entries(policy.max_entries, memory_type) + return + + for memory_lane in {entry.memory_type for entry in self._entries}: + self._enforce_max_entries(policy.max_entries, memory_lane) + + def _enforce_max_entries(self, max_entries: int, memory_type: MemoryType) -> None: + lane_entries = self.get_entries(memory_type) + overflow = len(lane_entries) - max_entries + if overflow <= 0: + return + keep_ids = {id(entry) for entry in lane_entries[-max_entries:]} + self._entries = [ + entry + for entry in self._entries + if entry.memory_type != memory_type or id(entry) in keep_ids + ] + + +__all__ = [ + "MemoryBackendConfig", + "MemoryEntry", + "MemoryLayerConfig", + "MemoryRetentionPolicy", + "MemoryStore", + "MemoryType", + "MCPServerMemoryProfile", +] diff --git a/scripts/coding/ai/mcp/registry.py b/scripts/coding/ai/mcp/registry.py new file mode 100644 index 00000000..9609e5ac --- /dev/null +++ b/scripts/coding/ai/mcp/registry.py @@ -0,0 +1,247 @@ +"""Declarative registry for Model Context Protocol servers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, Mapping, Tuple + +from .memory import ( + MCPServerMemoryProfile, + MemoryBackendConfig, + MemoryLayerConfig, + MemoryRetentionPolicy, + MemoryType, +) + + +@dataclass(frozen=True) +class RemoteMCPServer: + """Metadata to connect to a remote MCP server exposed over HTTP(S).""" + + name: str + url: str + mode: str = "readonly" + headers: Mapping[str, str] = field(default_factory=dict) + + def as_cli_entry(self) -> Dict[str, object]: + payload: Dict[str, object] = {"url": self.url, "mode": self.mode} + if self.headers: + payload["headers"] = dict(self.headers) + return payload + + +@dataclass(frozen=True) +class LocalMCPServer: + """Metadata for spawning a local MCP server via CLI (e.g., Playwright).""" + + name: str + command: str + args: Tuple[str, ...] + allowed_origins: Tuple[str, ...] + output_dir: str + viewport_size: Tuple[int, int] + env: Mapping[str, str] = field(default_factory=dict) + + def as_cli_entry(self) -> Dict[str, object]: + payload: Dict[str, object] = { + "command": self.command, + "args": list(self.args), + "metadata": { + "allowed_origins": list(self.allowed_origins), + "output_dir": self.output_dir, + "viewport_size": self.viewport_size, + }, + } + if self.env: + payload["env"] = dict(self.env) + return payload + + +@dataclass(frozen=True) +class MCPRegistry: + """Collection of remote/local MCP server definitions plus memory profiles.""" + + remote_servers: Tuple[RemoteMCPServer, ...] + local_servers: Tuple[LocalMCPServer, ...] + memory_profiles: Mapping[str, MCPServerMemoryProfile] = field(default_factory=dict) + + def as_cli_config(self) -> Dict[str, Dict[str, object]]: + """Serialize registry so CLI tooling can consume it directly.""" + + remote = {server.name: server.as_cli_entry() for server in self.remote_servers} + local = {server.name: server.as_cli_entry() for server in self.local_servers} + payload: Dict[str, Dict[str, object]] = {"remote": remote, "local": local} + if self.memory_profiles: + payload["memory_profiles"] = { + name: profile.as_dict() for name, profile in self.memory_profiles.items() + } + return payload + + +def build_default_registry() -> MCPRegistry: + """Build the registry that mirrors the Copilot CLI log shared by the user.""" + + remote_servers = ( + RemoteMCPServer( + name="blackbird-mcp-server", + url="https://api.githubcopilot.com/mcp/readonly", + mode="readonly", + ), + RemoteMCPServer( + name="github-mcp-server", + url="https://api.githubcopilot.com/mcp/readonly", + mode="readonly", + ), + ) + + allowed_origins = ("localhost", "localhost:*", "127.0.0.1", "127.0.0.1:*") + allowed_origins_flag = ";".join(allowed_origins) + viewport_size = (1280, 720) + + local_servers = ( + LocalMCPServer( + name="playwright", + command="npx", + args=( + "@playwright/mcp@0.0.40", + "--viewport-size", + f"{viewport_size[0]},{viewport_size[1]}", + "--output-dir", + "/tmp/playwright-logs", + "--allowed-origins", + allowed_origins_flag, + ), + allowed_origins=allowed_origins, + output_dir="/tmp/playwright-logs", + viewport_size=viewport_size, + ), + ) + + memory_profiles = _build_default_memory_profiles() + + return MCPRegistry( + remote_servers=remote_servers, + local_servers=local_servers, + memory_profiles=memory_profiles, + ) + + +def _build_default_memory_profiles() -> Dict[str, MCPServerMemoryProfile]: + """Mirror the agent memory stack so MCP servers feel identical.""" + + working_layer = MemoryLayerConfig( + memory_type=MemoryType.WORKING, + retention=MemoryRetentionPolicy(max_entries=5, ttl_seconds=900), + storage=MemoryBackendConfig( + provider="working_buffer", + description="Scratchpad that extracts requirements, proposals, and current asks.", + capabilities=("cot", "context-tracking"), + ), + purpose="Capture the most relevant snippets for the current MCP exchange.", + tools=("structured_note_taker",), + ) + + short_term_layer = MemoryLayerConfig( + memory_type=MemoryType.SHORT_TERM, + retention=MemoryRetentionPolicy(max_entries=20, ttl_seconds=7200), + storage=MemoryBackendConfig( + provider="conversation_buffer", + description="Stores the current MCP session so follow-ups keep the context.", + capabilities=("session-context", "followup-linking"), + ), + purpose="Maintain rolling conversational context for the duration of a workspace run.", + tools=("mem0",), + ) + + long_term_layer = MemoryLayerConfig( + memory_type=MemoryType.LONG_TERM, + retention=MemoryRetentionPolicy(max_entries=200), + storage=MemoryBackendConfig( + provider="mem0", + description="Persist user preferences, past actions, and lessons learned.", + capabilities=("preference-storage", "self-improvement"), + parameters={"workspace": "codex"}, + ), + purpose="Personalize the MCP experience across sessions and repos.", + tools=("mem0", "whiteboard_memory"), + ) + + persona_layer = MemoryLayerConfig( + memory_type=MemoryType.PERSONA, + retention=MemoryRetentionPolicy(max_entries=50), + storage=MemoryBackendConfig( + provider="whiteboard_memory", + description="Enforces the target persona (designer, developer, reviewer, etc.).", + capabilities=("role-grounding",), + ), + purpose="Keep the MCP responses aligned with the intended role/voice.", + tools=("persona_enforcer",), + ) + + episodic_layer = MemoryLayerConfig( + memory_type=MemoryType.EPISODIC, + retention=MemoryRetentionPolicy(max_entries=100), + storage=MemoryBackendConfig( + provider="mem0", + description="Records task-level attempts, failures, and outcomes.", + capabilities=("episode-tracking", "workflow-replay"), + ), + purpose="Let agents learn from past MCP workflows and improve autonomously.", + tools=("knowledge_agent",), + ) + + entity_layer = MemoryLayerConfig( + memory_type=MemoryType.ENTITY, + retention=MemoryRetentionPolicy(max_entries=500), + storage=MemoryBackendConfig( + provider="azure_ai_search", + description="Structured store for entities and relationships via Azure AI Search.", + capabilities=("entity-resolution", "structured-rag"), + parameters={"index": "codex-mcp-entities"}, + ), + purpose="Track people, repos, and deliverables mentioned during MCP activity.", + tools=("azure_ai_search",), + ) + + structured_rag_backend = MemoryBackendConfig( + provider="azure_ai_search", + description="Structured RAG profile optimized for multi-turn MCP workflows.", + capabilities=("structured-rag", "precise-grounding"), + parameters={"index": "codex-mcp-knowledge", "profile": "default"}, + ) + + github_profile = MCPServerMemoryProfile( + layers=( + working_layer, + short_term_layer, + long_term_layer, + persona_layer, + episodic_layer, + entity_layer, + ), + structured_rag=structured_rag_backend, + ) + + playwright_profile = MCPServerMemoryProfile( + layers=( + working_layer, + MemoryLayerConfig( + memory_type=MemoryType.SHORT_TERM, + retention=MemoryRetentionPolicy(max_entries=10, ttl_seconds=3600), + storage=MemoryBackendConfig( + provider="execution_buffer", + description="Short-term steps from Playwright E2E automation.", + capabilities=("workflow-steps",), + ), + purpose="Keep DOM + navigation context between MCP UI actions.", + tools=("playwright",), + ), + entity_layer, + ), + structured_rag=None, + ) + + return { + "github-mcp-server": github_profile, + "playwright": playwright_profile, + } diff --git a/scripts/coding/tests/ai/mcp/__init__.py b/scripts/coding/tests/ai/mcp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/scripts/coding/tests/ai/mcp/test_memory.py b/scripts/coding/tests/ai/mcp/test_memory.py new file mode 100644 index 00000000..0394710a --- /dev/null +++ b/scripts/coding/tests/ai/mcp/test_memory.py @@ -0,0 +1,97 @@ +"""Tests for MCP memory primitives and retention policies.""" + +import importlib.machinery +import sys +from pathlib import Path +from types import ModuleType + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[5] +SCRIPTS_ROOT = PROJECT_ROOT / "scripts" +CODING_ROOT = SCRIPTS_ROOT / "coding" + +sys.path.insert(0, str(PROJECT_ROOT)) +sys.path.insert(0, str(SCRIPTS_ROOT)) +sys.path.insert(0, str(CODING_ROOT)) + +namespace_paths = [str(SCRIPTS_ROOT), str(CODING_ROOT)] +scripts_pkg = ModuleType("scripts") +scripts_pkg.__package__ = "scripts" +scripts_pkg.__path__ = namespace_paths +scripts_pkg.__spec__ = importlib.machinery.ModuleSpec( + name="scripts", + loader=None, + is_package=True, +) +sys.modules.setdefault("scripts", scripts_pkg) + + +@pytest.fixture(name="memory_module") +def _memory_module(): + return __import__("scripts.coding.ai.mcp.memory", fromlist=["*"]) + + +def test_memory_store_groups_entries_by_type(memory_module): + store = memory_module.MemoryStore() + + working_entry = memory_module.MemoryEntry( + memory_type=memory_module.MemoryType.WORKING, + content="Book a trip to Paris", + metadata={"intent": "travel"}, + ) + long_term_entry = memory_module.MemoryEntry( + memory_type=memory_module.MemoryType.LONG_TERM, + content="Ben enjoys skiing and coffee with a mountain view", + metadata={"user": "Ben"}, + ) + + store.add_entry(working_entry) + store.add_entry(long_term_entry) + + working_entries = store.get_entries(memory_module.MemoryType.WORKING) + long_term_entries = store.get_entries(memory_module.MemoryType.LONG_TERM) + + assert working_entries == [working_entry] + assert long_term_entries == [long_term_entry] + + +def test_memory_store_prunes_entries_with_retention_policy(memory_module): + store = memory_module.MemoryStore() + policy = memory_module.MemoryRetentionPolicy(max_entries=2, ttl_seconds=60) + + first = memory_module.MemoryEntry( + memory_type=memory_module.MemoryType.SHORT_TERM, + content="Flight preference: window seat", + ) + second = memory_module.MemoryEntry( + memory_type=memory_module.MemoryType.SHORT_TERM, + content="Hotel preference: near museums", + ) + third = memory_module.MemoryEntry( + memory_type=memory_module.MemoryType.SHORT_TERM, + content="Restaurant preference: vegetarian", + ) + + store.add_entry(first) + store.add_entry(second) + store.add_entry(third) + + store.prune(policy, memory_type=memory_module.MemoryType.SHORT_TERM) + + remaining = store.get_entries(memory_module.MemoryType.SHORT_TERM) + assert len(remaining) == 2 + assert remaining[-1].content == "Restaurant preference: vegetarian" + + +def test_registry_exposes_memory_profiles(memory_module): + registry_module = __import__("scripts.coding.ai.mcp.registry", fromlist=["*"]) + registry = registry_module.build_default_registry() + + assert "github-mcp-server" in registry.memory_profiles + profile = registry.memory_profiles["github-mcp-server"] + + layers = {layer.memory_type: layer for layer in profile.layers} + assert memory_module.MemoryType.WORKING in layers + assert layers[memory_module.MemoryType.LONG_TERM].storage.provider == "mem0" + assert profile.structured_rag.provider == "azure_ai_search" diff --git a/scripts/coding/tests/ai/mcp/test_registry.py b/scripts/coding/tests/ai/mcp/test_registry.py new file mode 100644 index 00000000..591af903 --- /dev/null +++ b/scripts/coding/tests/ai/mcp/test_registry.py @@ -0,0 +1,68 @@ +"""Tests for the MCP registry builder mirroring the Copilot log.""" + +import importlib.machinery +import sys +from pathlib import Path +from types import ModuleType + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parents[5] +SCRIPTS_ROOT = PROJECT_ROOT / "scripts" +CODING_ROOT = SCRIPTS_ROOT / "coding" + +sys.path.insert(0, str(PROJECT_ROOT)) +sys.path.insert(0, str(SCRIPTS_ROOT)) +sys.path.insert(0, str(CODING_ROOT)) + +namespace_paths = [str(SCRIPTS_ROOT), str(CODING_ROOT)] +scripts_pkg = ModuleType("scripts") +scripts_pkg.__package__ = "scripts" +scripts_pkg.__path__ = namespace_paths +scripts_pkg.__spec__ = importlib.machinery.ModuleSpec( + name="scripts", + loader=None, + is_package=True, +) +sys.modules.setdefault("scripts", scripts_pkg) + + +@pytest.fixture(name="registry_module") +def _registry_module(): + return __import__("scripts.coding.ai.mcp.registry", fromlist=["*"]) + + +def test_default_registry_declares_remote_servers(registry_module): + registry = registry_module.build_default_registry() + + remote_names = [server.name for server in registry.remote_servers] + assert remote_names == ["blackbird-mcp-server", "github-mcp-server"] + assert {server.url for server in registry.remote_servers} == { + "https://api.githubcopilot.com/mcp/readonly" + } + assert all(server.mode == "readonly" for server in registry.remote_servers) + + +def test_playwright_server_configuration_matches_reference_log(registry_module): + registry = registry_module.build_default_registry() + local_servers = {server.name: server for server in registry.local_servers} + playwright = local_servers["playwright"] + + assert playwright.command == "npx" + assert playwright.viewport_size == (1280, 720) + assert playwright.output_dir == "/tmp/playwright-logs" + assert "@playwright/mcp@0.0.40" in playwright.args + assert "--allowed-origins" in playwright.args + assert "localhost;localhost:*;127.0.0.1;127.0.0.1:*" in playwright.args + + +def test_registry_cli_config_serializes_servers(registry_module): + registry = registry_module.build_default_registry() + cli_config = registry.as_cli_config() + + assert "remote" in cli_config and "local" in cli_config + assert cli_config["remote"]["github-mcp-server"]["url"].startswith("https://") + assert cli_config["local"]["playwright"]["command"] == "npx" + assert cli_config["local"]["playwright"]["args"][0] == "@playwright/mcp@0.0.40" + assert "memory_profiles" in cli_config + assert "github-mcp-server" in cli_config["memory_profiles"]