"""MCP Tool definitions for the Memory Layer.
This module consolidates write operations for the Memory layer as MCP Tools.
These tools allow the agent to record decisions, learnings, and report snapshots.
"""
from typing import Any
from mcp.server import Server
from .report_history import store_report
from .decisions import record_decision, get_decisions
from .learnings import add_learning
from .database import get_db_connection
def register_memory_tools(server: Server) -> None:
"""Register all memory tools with the MCP server.
These tools handle write operations to the memory layer:
- save_report_snapshot: Save performance reports for future comparison
- log_decision: Record decisions with rationale
- update_decision_outcome: Update the outcome of a past decision
- record_learning: Store insights and learnings
Args:
server: The MCP server instance.
"""
@server.tool()
async def save_report_snapshot(
report_type: str,
date_range: str,
data: dict[str, Any],
summary: str | None = None,
) -> dict[str, Any]:
"""Save a report snapshot for future comparison and trend analysis.
Use this tool after pulling performance reports to save them for historical
comparison. This enables week-over-week and month-over-month analysis.
Args:
report_type: Type of report (campaign, keyword, search_terms, performance).
date_range: The date range the report covers (e.g., "2024-01-01 to 2024-01-07").
data: The report data to save.
summary: Optional human-readable summary of key findings.
Returns:
Dictionary with the saved report ID and confirmation.
"""
report_id = store_report(
report_type=report_type,
date_range=date_range,
data=data,
summary=summary,
)
return {
"status": "saved",
"report_id": report_id,
"report_type": report_type,
"date_range": date_range,
}
@server.tool()
async def log_decision(
decision_type: str,
description: str,
rationale: str,
alternatives_considered: list[str] | None = None,
) -> dict[str, Any]:
"""Record a decision made about the Google Ads account.
Use this tool to document significant decisions such as strategy changes,
budget adjustments, keyword additions, or bid modifications. This creates
an audit trail and helps with future analysis.
Args:
decision_type: Type of decision. Common values:
- "strategy": High-level campaign strategy decisions
- "budget": Budget allocation or changes
- "targeting": Audience or location targeting changes
- "creative": Ad copy or creative decisions
- "bid": Bidding strategy or bid adjustments
- "keyword": Keyword additions, removals, or match type changes
description: What was decided.
rationale: Why this decision was made (data-driven reasoning).
alternatives_considered: Other options that were considered but rejected.
Returns:
Dictionary with the recorded decision ID.
"""
decision_id = record_decision(
decision_type=decision_type,
description=description,
rationale=rationale,
alternatives_considered=alternatives_considered,
)
return {
"status": "recorded",
"decision_id": decision_id,
"decision_type": decision_type,
}
@server.tool()
async def update_decision_outcome(
decision_id: int,
outcome: str,
) -> dict[str, Any]:
"""Update the outcome of a previously recorded decision.
Use this tool to close the loop on past decisions by recording their
actual results. This enables learning from what worked and what didn't.
Args:
decision_id: The ID of the decision to update (from log_decision).
outcome: The result/outcome of the decision. Include metrics if available
(e.g., "CTR improved 15% after 2 weeks" or "No significant change observed").
Returns:
Dictionary with update status.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE decisions SET outcome = ? WHERE id = ?",
(outcome, decision_id),
)
conn.commit()
if cursor.rowcount == 0:
return {"error": f"Decision {decision_id} not found"}
return {
"status": "updated",
"decision_id": decision_id,
"outcome": outcome,
}
@server.tool()
async def record_learning(
category: str,
insight: str,
evidence: str | None = None,
confidence: str = "medium",
actionable_recommendations: list[str] | None = None,
) -> dict[str, Any]:
"""Record a learning or insight about the Google Ads account.
Use this tool to capture insights discovered during analysis. These learnings
build institutional knowledge and inform future optimizations.
Args:
category: Category of learning. Common values:
- "audience": Insights about target audience behavior
- "keywords": Keyword performance patterns
- "creative": Ad copy or creative performance insights
- "bidding": Bid strategy effectiveness
- "timing": Day/time performance patterns
- "competitors": Competitive landscape observations
insight: The insight or learning (clear, actionable statement).
evidence: Data or observations supporting this insight.
confidence: Confidence level based on data strength:
- "low": Early signal, needs more data
- "medium": Reasonable evidence, worth acting on
- "high": Strong data support, high confidence
actionable_recommendations: Specific actions to take based on this learning.
Returns:
Dictionary with the recorded learning ID.
"""
if confidence not in ("low", "medium", "high"):
return {"error": "Confidence must be 'low', 'medium', or 'high'"}
learning_id = add_learning(
category=category,
insight=insight,
evidence=evidence,
confidence=confidence,
actionable_recommendations=actionable_recommendations,
)
return {
"status": "recorded",
"learning_id": learning_id,
"category": category,
"confidence": confidence,
}
