Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions packages/gg_api_core/src/gg_api_core/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1673,6 +1673,58 @@ async def delete_public_incident_note(self, incident_id: int | str, note_id: int
logger.info(f"Deleting note {note_id} from public incident {incident_id}")
return await self._request_delete(f"/public-incidents/secrets/{incident_id}/notes/{note_id}")

async def list_incident_activity_logs(
self,
incident_id: int | str,
params: dict[str, Any] | None = None,
get_all: bool = False,
) -> ListResponse:
"""List the full activity log (user notes AND system actions) of an internal secret incident.

Wraps GET /v1/incidents/secrets/{incident_id}/activity-logs.

Args:
incident_id: ID of the secret incident
params: Optional query parameters for filtering and pagination
get_all: If True, fetch all pages using paginate_all

Returns:
ListResponse with data, cursor, and has_more fields
"""
logger.info(f"Listing activity logs for incident {incident_id}")
endpoint = f"/incidents/secrets/{incident_id}/activity-logs"

if get_all:
return await self.paginate_all(endpoint, params)

return await self._request_list(endpoint, params=params)

async def list_public_incident_activity_logs(
self,
incident_id: int | str,
params: dict[str, Any] | None = None,
get_all: bool = False,
) -> ListResponse:
"""List the full activity log (user notes AND system actions) of a public secret incident.

Wraps GET /v1/public-incidents/secrets/{incident_id}/activity-logs.

Args:
incident_id: ID of the public secret incident
params: Optional query parameters for filtering and pagination
get_all: If True, fetch all pages using paginate_all

Returns:
ListResponse with data, cursor, and has_more fields
"""
logger.info(f"Listing activity logs for public incident {incident_id}")
endpoint = f"/public-incidents/secrets/{incident_id}/activity-logs"

if get_all:
return await self.paginate_all(endpoint, params)

return await self._request_list(endpoint, params=params)

# Secret Occurrences management
async def list_secret_occurrences(self, incident_id: str) -> dict[str, Any]:
"""List secret occurrences for an incident.
Expand Down
149 changes: 149 additions & 0 deletions packages/gg_api_core/src/gg_api_core/tools/activity_logs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
"""Tools for listing the activity log of a secret incident.

An incident's activity log is the full, chronological record of everything that
happened to it: user notes (free-form comments) AND system actions (status
changes, assignments, severity edits, new locations detected, etc.). The public
API exposes it under ``.../activity-logs`` and each entry carries a ``content``
discriminated union — ``{"type": "note", "comment": ...}`` for a comment or
``{"type": "action", "content_key": ..., "data": ...}`` for a system action.

This is broader than the notes-only listing (``list_incident_comments``): use
the activity log to reconstruct an incident's full timeline. Internal incidents
and Public Monitoring incidents have separate, non-interchangeable activity
logs, so each perimeter gets its own dedicated tool — mirroring the split used
elsewhere (e.g. ``assign_incident`` vs ``assign_public_incident``).
"""

import logging
from typing import Any

from fastmcp.exceptions import ToolError
from pydantic import BaseModel, Field

from gg_api_core.client import DEFAULT_PAGINATION_MAX_BYTES, ListResponse
from gg_api_core.utils import get_client

logger = logging.getLogger(__name__)


class ListActivityLogsParams(BaseModel):
"""Parameters for listing the activity log of a secret incident."""

incident_id: int = Field(description="ID of the secret incident whose activity log to list")
content_key: str | None = Field(
default=None,
description="Filter to a single system-action type (e.g. 'TRIGGER', 'RESOLVE', 'ASSIGN', 'SET_SEVERITY'). "
"Omit to return notes and every action type.",
)
member_id: int | None = Field(
default=None,
description="Filter to entries authored by a specific member (by member ID)",
)
cursor: str | None = Field(default=None, description="Pagination cursor for fetching the next page of results")
per_page: int = Field(default=20, description="Number of results per page (default: 20, min: 1, max: 100)")
get_all: bool = Field(
default=False,
description=f"If True, fetch all pages (capped at ~{DEFAULT_PAGINATION_MAX_BYTES / 1000}KB; check 'has_more' and use cursor to continue)",
)


class ListActivityLogsResult(BaseModel):
"""Result from listing the activity log of a secret incident."""

activity_logs: list[dict[str, Any]] = Field(
description="List of activity log entries (notes and system actions) attached to the incident"
)
total_count: int = Field(description="Total number of entries returned")
next_cursor: str | None = Field(default=None, description="Pagination cursor for next page (if applicable)")
has_more: bool = Field(default=False, description="True if more results exist (use next_cursor to fetch)")


def _build_query_params(params: ListActivityLogsParams) -> dict[str, Any]:
"""Build the query parameters dict from the tool params, dropping unset filters."""
query_params: dict[str, Any] = {"per_page": params.per_page}
if params.cursor:
query_params["cursor"] = params.cursor
if params.content_key:
query_params["content_key"] = params.content_key
if params.member_id is not None:
query_params["member_id"] = params.member_id
return query_params


def _build_list_result(result: ListResponse) -> ListActivityLogsResult:
"""Adapt a client ListResponse into a ListActivityLogsResult."""
return ListActivityLogsResult(
activity_logs=result["data"],
total_count=len(result["data"]),
next_cursor=result["cursor"],
has_more=result["has_more"],
)


async def list_incident_activity_logs(params: ListActivityLogsParams) -> ListActivityLogsResult:
"""
List the full activity log of an internal secret incident.

Returns both user notes (comments) and system actions (status changes,
assignments, severity edits, new locations detected, etc.) in one timeline.
Use this to reconstruct what happened to an incident and when. To read only
the comments use `list_incident_comments`; for Public Monitoring incidents
use `list_public_incident_activity_logs` instead.

Args:
params: ListActivityLogsParams with the incident ID, optional filters and pagination options

Returns:
ListActivityLogsResult with the entries, total_count, next_cursor, and has_more

Raises:
ToolError: If the listing operation fails
"""
client = await get_client()
logger.debug(f"Listing activity logs for incident {params.incident_id}")

try:
result = await client.list_incident_activity_logs(
incident_id=params.incident_id,
params=_build_query_params(params),
get_all=params.get_all,
)
return _build_list_result(result)
except Exception as e:
logger.exception(f"Error listing activity logs for incident {params.incident_id}: {str(e)}")
raise ToolError(f"Error: {str(e)}")


async def list_public_incident_activity_logs(params: ListActivityLogsParams) -> ListActivityLogsResult:
"""
List the full activity log of a public secret incident.

Public incidents are surfaced by GitGuardian Public Monitoring (public GitHub
repos/gists, Docker Hub, etc.). Returns both user notes (comments) and system
actions (status changes, assignments, severity edits, etc.) in one timeline.
To read only the comments use `list_public_incident_comments`; for internal
incidents use `list_incident_activity_logs` instead. Public incident IDs are
NOT interchangeable with internal incident IDs.

Args:
params: ListActivityLogsParams with the public incident ID, optional filters and pagination options

Returns:
ListActivityLogsResult with the entries, total_count, next_cursor, and has_more

Raises:
ToolError: If the listing operation fails
"""
client = await get_client()
logger.debug(f"Listing activity logs for public incident {params.incident_id}")

try:
result = await client.list_public_incident_activity_logs(
incident_id=params.incident_id,
params=_build_query_params(params),
get_all=params.get_all,
)
return _build_list_result(result)
except Exception as e:
logger.exception(f"Error listing activity logs for public incident {params.incident_id}: {str(e)}")
raise ToolError(f"Error: {str(e)}")
27 changes: 25 additions & 2 deletions packages/gg_mcp_server/src/gg_mcp_server/register_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,10 @@

from fastmcp.exceptions import ToolError
from gg_api_core.mcp_server import AbstractGitGuardianFastMCP
from gg_api_core.tools.activity_logs import (
list_incident_activity_logs,
list_public_incident_activity_logs,
)
from gg_api_core.tools.assign_incident import assign_incident
from gg_api_core.tools.assign_public_incident import assign_public_incident
from gg_api_core.tools.count_incidents import count_incidents
Expand Down Expand Up @@ -68,13 +72,14 @@
private/org Git repos, Slack, Jira, Confluence, container registries, SharePoint, etc.
Identified by a `source_id`. Default category for most customer workflows.
Read tools: `list_incidents`, `count_incidents`, `get_incident`, `list_repo_occurrences`,
`remediate_secret_incidents`, `list_sources`, `find_current_source_id`, `list_incident_comments`.
`remediate_secret_incidents`, `list_sources`, `find_current_source_id`, `list_incident_comments`,
`list_incident_activity_logs`.
Write tools: `manage_private_incident`, `update_incident_status`, `assign_incident`,
`update_or_create_incident_custom_tags`, `create_code_fix_request`, `manage_incident_comment`.
- **Public incidents** — detected by GitGuardian Public Monitoring on the worldwide public
perimeter: public GitHub repos/gists, Docker Hub, etc. Not linked to a workspace source.
Read tools: `list_public_incidents`, `get_public_incident`, `list_public_occurrences`,
`list_public_incident_comments`.
`list_public_incident_comments`, `list_public_incident_activity_logs`.
Write tools: `assign_public_incident`, `update_public_incident_status`, `manage_public_incident_comment`.

Incident IDs are **not** interchangeable between the two categories. If the user's intent is
Expand Down Expand Up @@ -290,6 +295,24 @@ def register_tools(mcp: AbstractGitGuardianFastMCP) -> None:
required_scopes=["incidents:read"],
)

mcp.tool(
list_incident_activity_logs,
description="(Internal sources only — for public GitHub/gists/Docker Hub use list_public_incident_activity_logs) "
"List the full activity log of an internal secret incident: both user notes (comments) and system actions "
"(status changes, assignments, severity edits, new locations detected, etc.) in one timeline. Broader than "
"list_incident_comments, which returns notes only. Filter by content_key (action type) or member_id.",
required_scopes=["incidents:read"],
)

mcp.tool(
list_public_incident_activity_logs,
description="(Public Monitoring only — for internal sources use list_incident_activity_logs) "
"List the full activity log of a public secret incident detected by GitGuardian Public Monitoring: both user "
"notes (comments) and system actions in one timeline. Broader than list_public_incident_comments, which "
"returns notes only. Public incident IDs are NOT interchangeable with internal incident IDs.",
required_scopes=["incidents:read"],
)

# Write tools — previously SecOps-only.

@mcp.tool(
Expand Down
79 changes: 79 additions & 0 deletions tests/cassettes/client/vcr/test_list_incident_activity_logs.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
interactions:
- request:
body: ''
headers:
Accept:
- '*/*'
Accept-Encoding:
- gzip, deflate
Connection:
- keep-alive
Content-Type:
- application/json
Host:
- api.gitguardian.com
User-Agent:
- GitGuardian-MCP-Server/0.5.0
X-Privacy-Mode:
- 'true'
method: GET
uri: https://api.gitguardian.com/v1/incidents/secrets/21460/activity-logs
response:
body:
string: '[{"id": 9001, "incident_id": 21460, "member": {"id": 480870, "name":
"Jane Doe", "email": "jane.doe@example.com", "access_level": "owner"}, "api_token_id":
null, "created_at": "2026-06-23T10:00:00.123456Z", "updated_at": null, "content":
{"type": "note", "comment": "Investigating this incident via MCP"}}, {"id":
9002, "incident_id": 21460, "member": {"id": 480870, "name": "Jane Doe", "email":
"jane.doe@example.com", "access_level": "owner"}, "api_token_id": null, "created_at":
"2026-06-23T10:05:00.654321Z", "updated_at": null, "content": {"type": "action",
"content_key": "RESOLVE", "data": {"resolution": "revoked"}}}]'
headers:
CF-RAY:
- a10417019cc9229d-CDG
Connection:
- keep-alive
Content-Type:
- application/json
Date:
- Tue, 23 Jun 2026 10:05:01 GMT
Server:
- cloudflare
Transfer-Encoding:
- chunked
access-control-expose-headers:
- X-App-Version
allow:
- GET, HEAD, OPTIONS
cf-cache-status:
- DYNAMIC
content-security-policy:
- frame-ancestors 'none'
cross-origin-opener-policy:
- same-origin
link:
- ''
referrer-policy:
- strict-origin-when-cross-origin
strict-transport-security:
- max-age=31536000; includeSubDomains
vary:
- Cookie
x-app-version:
- v2.521.0
x-content-type-options:
- nosniff
x-envoy-upstream-service-time:
- '28'
x-frame-options:
- DENY
x-per-page:
- '20'
x-secrets-engine-version:
- 2.165.0
x-xss-protection:
- 1; mode=block
status:
code: 200
message: OK
version: 1
Loading
Loading