get_aggregate_data
Fetch aggregate data reports from the AppsFlyer Pull API, including daily, geo, and partner metrics, by specifying app ID, date range, and report type.
Instructions
Fetches aggregate data reports from the AppsFlyer Pull API.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| data | Yes |
Input Schema (JSON Schema)
{
"$defs": {
"AggregateDataInput": {
"properties": {
"app_id": {
"description": "The ID of the AppsFlyer app.",
"title": "App Id",
"type": "string"
},
"from_date": {
"description": "The start date of the data range (YYYY-MM-DD).",
"format": "date",
"title": "From Date",
"type": "string"
},
"report_type": {
"default": "daily_report",
"description": "The type of aggregate report to fetch.",
"enum": [
"partners_report",
"partners_by_date_report",
"daily_report",
"geo_report",
"geo_by_date_report"
],
"title": "Report Type",
"type": "string"
},
"to_date": {
"description": "The end date of the data range (YYYY-MM-DD).",
"format": "date",
"title": "To Date",
"type": "string"
}
},
"required": [
"app_id",
"from_date",
"to_date"
],
"title": "AggregateDataInput",
"type": "object"
}
},
"properties": {
"data": {
"$ref": "#/$defs/AggregateDataInput"
}
},
"required": [
"data"
],
"title": "get_aggregate_dataArguments",
"type": "object"
}
Implementation Reference
- appsflyer_mcp/server.py:23-34 (schema)Pydantic BaseModel defining the input schema for the get_aggregate_data tool, including app_id, date range, and report_type.class AggregateDataInput(BaseModel): app_id: str = Field(..., description="The ID of the AppsFlyer app.") from_date: date = Field(..., description="The start date of the data range (YYYY-MM-DD).") to_date: date = Field(..., description="The end date of the data range (YYYY-MM-DD).") report_type: Literal[ "partners_report", "partners_by_date_report", "daily_report", "geo_report", "geo_by_date_report" ] = Field("daily_report", description="The type of aggregate report to fetch.")
- appsflyer_mcp/server.py:36-36 (registration)The @mcp.tool() decorator registers the get_aggregate_data function as an MCP tool on the FastMCP server instance.@mcp.tool()
- appsflyer_mcp/server.py:37-67 (handler)The core handler function that executes the tool logic: validates credentials, constructs the AppsFlyer API endpoint, sends an asynchronous HTTP GET request, and returns the CSV response or error message.async def get_aggregate_data(data: AggregateDataInput): """Fetches aggregate data reports from the AppsFlyer Pull API.""" if not AF_API_BASE_URL or not AF_TOKEN: return "Error: AppsFlyer API credentials not configured." # The specific endpoint for the AGGREGATE PULL API endpoint = f"{AF_API_BASE_URL}/api/agg-data/export/app/{data.app_id}/{data.report_type}/v5" params = { "from": data.from_date.isoformat(), "to": data.to_date.isoformat(), } headers = { "Authorization": f"Bearer {AF_TOKEN}", "Accept": "text/csv" } try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.get(endpoint, headers=headers, params=params) response.raise_for_status() return response.text except httpx.HTTPStatusError as e: return f"HTTP error occurred: {e.response.text}" except httpx.RequestError as e: return f"A network error occurred: {e}"