get_collection_timeline_events
Retrieve curated timeline events for security campaigns and threat actors from Google Threat Intelligence collections to analyze threat activity chronologically.
Instructions
Retrieves timeline events from the given collection, when available.
This is super valuable curated information produced by security analysits at Google Threat Intelligence.
We should fetch this information for campaigns and threat actors always.
It's common to display the events grouped by the "event_category" field.
Args: id (required): Collection identifier Return: List of events related to the given collection.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| id | Yes | ||
| api_key | No |
Implementation Reference
- gti_mcp/tools/collections.py:384-406 (handler)The main handler function get_collection_timeline_events that retrieves timeline events from a collection via VirusTotal API. Decorated with @server.tool() for MCP registration.
@server.tool() async def get_collection_timeline_events(id: str, ctx: Context, api_key: str = None): """Retrieves timeline events from the given collection, when available. This is super valuable curated information produced by security analysits at Google Threat Intelligence. We should fetch this information for campaigns and threat actors always. It's common to display the events grouped by the "event_category" field. Args: id (required): Collection identifier Return: List of events related to the given collection. """ async with vt_client(ctx, api_key=api_key) as client: resp = await client.get_async(f"/collections/{id}/timeline/events") if resp.status != 200: error_json = await resp.json_async() error_info = error_json.get("error", {}) return [{"error": f"API Error: {error_info.get('message', 'Unknown error')}"}] data = await resp.json_async() return utils.sanitize_response(data.get("data", [])) - gti_mcp/tools/collections.py:384-384 (registration)The @server.tool() decorator that registers get_collection_timeline_events as an MCP tool. The server instance is a FastMCP object from gti_mcp/server.py.
@server.tool() - gti_mcp/utils.py:119-138 (helper)Helper function sanitize_response that recursively removes empty dictionaries and lists from API responses, used by get_collection_timeline_events to clean up returned data.
def sanitize_response(data: typing.Any) -> typing.Any: """Removes empty dictionaries and lists recursively from a response.""" if isinstance(data, dict): sanitized_dict = {} for key, value in data.items(): sanitized_value = sanitize_response(value) if sanitized_value is not None: sanitized_dict[key] = sanitized_value return sanitized_dict elif isinstance(data, list): sanitized_list = [] for item in data: sanitized_item = sanitize_response(item) if sanitized_item is not None: sanitized_list.append(sanitized_item) return sanitized_list elif isinstance(data, str): return data if data else None else: return data - Test case for get_collection_timeline_events that verifies error handling when the API returns a non-200 status code (e.g., 404 for non-existent collection).
async def test_get_collection_timeline_events_api_error_handled(): """Test get_collection_timeline_events returns error dict on API error (e.g. 404).""" mock_ctx = MagicMock(spec=Context) mock_response = MagicMock() mock_response.status = 404 async def json_async(): return {"error": {"message": "Collection not found"}} mock_response.json_async = json_async mock_client_instance = MagicMock() mock_client_instance.get_async = AsyncMock(return_value=mock_response) mock_vt_client = MagicMock() mock_vt_client.__aenter__ = AsyncMock(return_value=mock_client_instance) mock_vt_client.__aexit__ = AsyncMock(return_value=None) with patch("gti_mcp.tools.collections.vt_client", return_value=mock_vt_client): result = await collections.get_collection_timeline_events(id="non_existent_id", ctx=mock_ctx) assert isinstance(result, list) assert len(result) == 1 assert "error" in result[0] assert "Collection not found" in result[0]["error"]