"""MCP Resource definitions for the Memory Layer.
This module registers read-only Memory resources with the MCP server.
These resources provide access to historical data, change logs, decisions,
and accumulated learnings.
"""
import json
from typing import Any
from mcp.server import Server
from .change_log import get_recent_changes
from .report_history import get_report_history
from .decisions import get_decisions
from .learnings import get_learnings
def get_memory_resources() -> list[dict[str, Any]]:
"""Return the list of available memory resources.
Returns:
List of resource definitions with uri, name, description, and mimeType.
"""
return [
{
"uri": "memory://changes/recent",
"name": "Recent Changes",
"description": "Recent change log entries showing all modifications made to the Google Ads account",
"mimeType": "application/json",
},
{
"uri": "memory://reports/history",
"name": "Report History",
"description": "Historical report snapshots saved for trend analysis and comparison",
"mimeType": "application/json",
},
{
"uri": "memory://decisions/history",
"name": "Decision History",
"description": "Past decisions with rationale, alternatives considered, and outcomes",
"mimeType": "application/json",
},
{
"uri": "memory://insights",
"name": "Insights",
"description": "Accumulated learnings and insights from account performance analysis",
"mimeType": "application/json",
},
{
"uri": "memory://insights/actionable",
"name": "Actionable Insights",
"description": "High-confidence insights with specific actionable recommendations",
"mimeType": "application/json",
},
]
def read_memory_resource(uri: str) -> str:
"""Read a memory resource by URI.
Args:
uri: The resource URI to read.
Returns:
JSON-serialized content of the resource.
Raises:
ValueError: If the URI is not recognized.
"""
if uri == "memory://changes/recent":
data = get_recent_changes(limit=50)
content = {"changes": data, "count": len(data)}
elif uri == "memory://reports/history":
data = get_report_history(limit=20)
content = {"reports": data, "count": len(data)}
elif uri == "memory://decisions/history":
data = get_decisions(limit=20)
content = {"decisions": data, "count": len(data)}
elif uri == "memory://insights":
data = get_learnings(limit=30)
content = {"learnings": data, "count": len(data)}
elif uri == "memory://insights/actionable":
# Get high confidence learnings that have recommendations
data = get_learnings(confidence="high", limit=20)
actionable = [l for l in data if l.get("actionable_recommendations")][:10]
content = {"insights": actionable, "count": len(actionable)}
else:
raise ValueError(f"Unknown memory resource URI: {uri}")
return json.dumps(content, indent=2, default=str)
def register_memory_resources(server: Server) -> None:
"""Register all memory resources with the MCP server.
Note: MCP only allows one @server.list_resources() handler globally.
This function exports get_memory_resources() and read_memory_resource()
which will be called from a unified handler in server.py (Phase 3).
Args:
server: The MCP server instance.
"""
# The actual handler registration will happen in Phase 3 when we
# combine knowledge and memory resources in server.py.
# For now, this module provides the data access functions.
pass
