Forecast commentary agents that act on-behalf-of your RevOps team

query = (
    "SELECT Id, Name, StageName, Amount, CloseDate, IsClosed "
    "FROM Opportunity WHERE IsClosed = false "
    "ORDER BY CloseDate ASC LIMIT 2000"
)
sf_records = self.execute_tool(
    "salesforce_soql_execute", soql_query=query,
).get("records") or []

# HubSpot: resolve open stage IDs first, then search deals
open_stage_labels = resolve_hubspot_open_stages(hubspot)
# -> hubspot_deal_pipelines_list(objectType="deals")

data = self.execute_tool(
    "hubspot_deals_search",
    filterGroups=[{"filters": [{
        "propertyName": "dealstage",
        "operator": "IN",
        "values": open_stage_ids,
    }]}],
    properties=["dealname", "amount", "dealstage",
                "closedate", "pipeline"],
    limit=limit, offset=offset,
)

def calculate_coverage_ratio(total_open_value, quota_target):
    """coverage_ratio = total_open_pipeline_value / quota_target"""
    if quota_target <= 0:
        return 0.0
    return round(total_open_value / quota_target, 2)

# At-risk signals, per stage:
#   stale -- open deals whose close date already passed
#   thin  -- a late stage with fewer than 2 deals
#   light -- a late stage holding < 5% of total pipeline value
if segment.overdue_count() > 0:
    reasons.append(
        f"{segment.overdue_count()} of {segment.deal_count} "
        f"deal(s) have a close date in the past"
    )

def compute_pipeline_fingerprint(segments, at_risk_flags):
    payload = {
        "segments": {
            label: {
                "deal_count": segment.deal_count,
                "total_value": round(segment.total_value),
                "sources": sorted(segment.sources.keys()),
            }
            for label, segment in sorted(segments.items())
        },
        "at_risk": sorted(at_risk_flags.keys()),
    }
    blob = json.dumps(payload, sort_keys=True)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

if not state.has_changed(cfg.analyst_email, fingerprint):
    logger.info(
        "Pipeline unchanged since the last Slack post "
        "-- skipping Slack"
    )

commentary = draft_commentary(
    segments=segments, at_risk_flags=at_risk_flags,
    total_open_value=total_open_value,
    coverage_ratio=coverage_ratio, ...
)
commentary = polish_commentary(commentary, ...)  # optional LLM pass

channel_id = slack.resolve_channel_id(cfg.slack_channel)
# "#revenue-ops" -> slackmcp_slack_search_channels -> "C..."

def send_message(self, channel_id, text):
    return self.execute_tool(
        "slackmcp_slack_send_message",
        channel_id=channel_id,
        message=text,
    )

sheets.ensure_tab(spreadsheet_id, tab_name)
# -> googlesheets_add_sheet if the tab doesn't exist yet

for label, segment in segments.items():
    row = [run_date, cfg.analyst_email, cfg.forecast_period,
           label, source_str, segment.deal_count,
           round(segment.total_value, 2), coverage_ratio,
           "TRUE" if label in at_risk_flags else "FALSE"]
    sheets.append_row(spreadsheet_id, tab_name, row)

def append_row(self, spreadsheet_id, tab_name, row):
    return self.execute_tool(
        "googlesheets_append_values",
        spreadsheet_id=spreadsheet_id,
        range=f"{tab_name}!A1",
        value_input_option="USER_ENTERED",  # snake_case, verified live
        values=[row],
    )