Platform Activity Log

Track and audit all workspace activities with comprehensive event logging and history

Activity Log provides comprehensive audit trail and event tracking for all workspace activities, enabling team transparency, debugging, and compliance monitoring.

Activity Log provides comprehensive audit trail and event tracking for all workspace activities, enabling team transparency, debugging, and compliance monitoring.

python
from praisonaiagents import Agent

agent = Agent(name="auditor", instructions="Explain recent workspace activity.")
agent.start("Summarise activity on this workspace in the last hour.")

The user reviews the audit trail to see who changed issues, agents, or membership and when.

Diagram

Activity Tracking

⚡ Event Occurs

📝 Capture Details

💾 Store Activity

👁️ View History

Quick Start

  1. View Recent Activity
    python
    import asyncio
    from praisonai_platform.client import PlatformClient
    
    async def view_activity():
        client = PlatformClient("http://localhost:8000", token="your-jwt-token")
        ws_id = "your-workspace-id"
    
        # Get recent workspace activity
        activities = await client.get_workspace_activity(
            ws_id,
            limit=50,  # Last 50 activities
            include_system=True  # Include system events
        )
    
        for activity in activities:
            print(f"[{activity['timestamp']}] {activity['actor']['name']}")
            print(f"  {activity['action']} {activity['resource_type']}: {activity['description']}")
            print()
    
    asyncio.run(view_activity())
  2. Filter Activity by Type
    python
    async def filter_activity():
        client = PlatformClient("http://localhost:8000", token="your-jwt-token")
        ws_id = "your-workspace-id"
    
        # Filter by specific activity types
        issue_activities = await client.get_workspace_activity(
            ws_id,
            resource_types=["issue"],
            actions=["created", "updated", "assigned"],
            start_date="2025-01-01T00:00:00Z",
            end_date="2025-01-31T23:59:59Z"
        )
    
        print(f"Found {len(issue_activities)} issue activities in January")
    
        # Filter by specific user
        user_activities = await client.get_workspace_activity(
            ws_id,
            actor_id="user-123",
            limit=25
        )
    
        print(f"User performed {len(user_activities)} activities")
    
    asyncio.run(filter_activity())

How It Works

Diagram
DatabaseEvent QueueActivity ServicePlatform APIUserDatabaseEvent QueueActivity ServicePlatform APIUserPerform ActionEmit EventQueue ActivityStore Activity LogConfirm StoredRequest Activity HistoryQuery ActivitiesFetch ActivitiesActivity DataActivity ListActivity History
ComponentPurposeData Captured
Activity EventSingle workspace actionActor, timestamp, action, resource
Resource ContextWhat was affectedResource type, ID, name, metadata
Actor InformationWho performed actionUser ID, name, role, IP address
Change DetailsWhat changedBefore/after values, diff summary

Activity Types

Core Platform Events

ActionResource TypeDescriptionExample
createdissue, project, labelResource creation“Created issue ISS-123”
updatedissue, project, userResource modification“Updated issue status to ‘done'”
deletedissue, label, commentResource deletion“Deleted comment on ISS-123”
assignedissueAssignment change“Assigned issue to @john”

Workspace Events

ActionResource TypeDescriptionExample
member_addedworkspaceNew member joined“Added @sarah to workspace”
member_removedworkspaceMember removed“Removed @john from workspace”
role_changedworkspace_memberPermission change“Changed @sarah role to admin”
settings_updatedworkspaceWorkspace configuration“Updated workspace settings”

Agent Events

ActionResource TypeDescriptionExample
agent_assignedissueAgent assignment“Assigned AI agent to ISS-123”
agent_completedissueAgent task completion“Agent completed issue analysis”
agent_failedissueAgent task failure“Agent failed to process issue”
agent_createdagentNew agent created“Created agent ‘Code Reviewer'”

API Reference

Activity Endpoints

MethodEndpointPurposeAuthentication
GET/api/v1/workspaces/{ws_id}/activityList workspace activityBearer Token
GET/api/v1/workspaces/{ws_id}/activity/{activity_id}Get activity detailsBearer Token
GET/api/v1/issues/{issue_id}/activityList issue-specific activityBearer Token
GET/api/v1/users/me/activityList user’s activity across workspacesBearer Token

Query Parameters

ParameterTypeDescriptionExample
limitintegerMax activities to return?limit=50
offsetintegerSkip activities for pagination?offset=100
actor_idstringFilter by specific user?actor_id=user-123
resource_typesstring[]Filter by resource types?resource_types=issue,project
actionsstring[]Filter by action types?actions=created,updated
start_dateISO dateActivity after date?start_date=2025-01-01T00:00:00Z
end_dateISO dateActivity before date?end_date=2025-01-31T23:59:59Z
include_systembooleanInclude system events?include_system=true

Common Patterns

Activity Feed Dashboard

Build a real-time activity dashboard for team visibility:

python
async def activity_dashboard():
    client = PlatformClient("http://localhost:8000", token="your-jwt-token")
    ws_id = "your-workspace-id"

    # Get different activity types for dashboard
    recent_activity = await client.get_workspace_activity(ws_id, limit=20)
    issue_activity = await client.get_workspace_activity(
        ws_id, 
        resource_types=["issue"], 
        limit=10
    )
    member_activity = await client.get_workspace_activity(
        ws_id,
        actions=["member_added", "member_removed", "role_changed"],
        limit=5
    )

    # Format for display
    dashboard_data = {
        "recent_activities": recent_activity,
        "issue_updates": issue_activity,
        "team_changes": member_activity,
        "summary": {
            "total_activities_today": len([
                a for a in recent_activity 
                if a['timestamp'].startswith('2025-01-15')  # Today
            ]),
            "active_users": len(set(a['actor']['id'] for a in recent_activity)),
            "issues_updated": len([
                a for a in issue_activity 
                if a['action'] == 'updated'
            ])
        }
    }

    return dashboard_data
Audit Trail Export

Export activity logs for compliance and audit purposes:

python
async def export_audit_trail():
    client = PlatformClient("http://localhost:8000", token="your-jwt-token")
    ws_id = "your-workspace-id"

    # Export all activities for a date range
    import csv
    from datetime import datetime, timedelta, timezone

    # Get last 30 days of activity
    end_date = datetime.now(timezone.utc)
    start_date = end_date - timedelta(days=30)

    all_activities = []
    offset = 0
    limit = 100

    while True:
        activities = await client.get_workspace_activity(
            ws_id,
            start_date=start_date.isoformat(),
            end_date=end_date.isoformat(),
            offset=offset,
            limit=limit,
            include_system=True
        )

        if not activities:
            break

        all_activities.extend(activities)
        offset += limit

    # Export to CSV
    with open('audit_trail.csv', 'w', newline='') as csvfile:
        fieldnames = ['timestamp', 'actor', 'action', 'resource_type', 'resource_id', 'description']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for activity in all_activities:
            writer.writerow({
                'timestamp': activity['timestamp'],
                'actor': activity['actor']['name'],
                'action': activity['action'],
                'resource_type': activity['resource_type'],
                'resource_id': activity['resource_id'],
                'description': activity['description']
            })

    print(f"Exported {len(all_activities)} activities to audit_trail.csv")
Real-time Activity Monitoring

Monitor workspace activity in real-time using webhooks:

python
async def monitor_workspace_activity():
    client = PlatformClient("http://localhost:8000", token="your-jwt-token")
    ws_id = "your-workspace-id"

    # Set up webhook for real-time activity notifications
    webhook_config = {
        "url": "https://your-app.com/webhooks/activity",
        "events": ["activity.created"],
        "filters": {
            "workspace_id": ws_id,
            "resource_types": ["issue", "project"],
            "actions": ["created", "updated", "assigned"]
        }
    }

    webhook = await client.create_webhook(ws_id, webhook_config)
    print(f"Created webhook: {webhook['id']}")

    # Alternative: Long-polling for activity updates
    last_activity_id = None
    while True:
        activities = await client.get_workspace_activity(
            ws_id,
            limit=10,
            after_id=last_activity_id  # Only get new activities
        )

        for activity in activities:
            print(f"New activity: {activity['description']}")
            last_activity_id = activity['id']

            # Process activity (send notifications, update caches, etc.)
            await process_activity_notification(activity)

        # Wait before checking again
        await asyncio.sleep(5)

async def process_activity_notification(activity):
    """Process new activity for notifications or integrations"""
    if activity['action'] == 'created' and activity['resource_type'] == 'issue':
        # Notify team of new issue
        print(f"🆕 New issue created: {activity['description']}")
    elif activity['action'] == 'assigned':
        # Notify assigned user
        print(f"📋 Issue assigned: {activity['description']}")

Best Practices

Performance Optimization
  • Pagination: Always use pagination for large activity queries
  • Date filtering: Use specific date ranges to reduce query scope
  • Resource filtering: Filter by specific resource types when possible
  • Caching: Cache frequently accessed activity data
Privacy and Security
  • Access control: Ensure users only see activities they have permission for
  • Sensitive data: Avoid logging sensitive information in activity descriptions
  • Retention policies: Implement data retention policies for old activities
  • Audit compliance: Maintain activity logs for required compliance periods
User Experience
  • Meaningful descriptions: Use clear, human-readable activity descriptions
  • Relevant context: Include enough context to understand the activity
  • Grouping: Group related activities to reduce noise
  • Real-time updates: Provide real-time activity feeds where appropriate
Integration Patterns
  • Webhook integration: Use webhooks for real-time external integrations
  • Batch processing: Process activities in batches for performance
  • Event sourcing: Consider activity logs as event source for state reconstruction
  • Analytics: Aggregate activity data for workspace insights and metrics

Related