check_entity_updates

Identify modified entities in Kanka campaigns by comparing current data against a specified sync timestamp to maintain accurate campaign information.

Instructions

Check which entity_ids have been modified since last sync

Input Schema

TableJSON Schema

Name	Required	Description	Default
`entity_ids`	Yes	Array of entity IDs to check
`last_synced`	Yes	ISO 8601 timestamp to check updates since

Implementation Reference

src/mcp_kanka/operations.py:694-755 (handler)

The core implementation of check_entity_updates - fetches all entities from Kanka API, compares their updated_at timestamps with last_synced parameter, and returns lists of modified and deleted entity IDs with a check timestamp.

async def check_entity_updates(
    self, entity_ids: list[int], last_synced: str
) -> CheckEntityUpdatesResult:
    """Check which entities have been modified since last sync.

    Args:
        entity_ids: List of entity IDs to check
        last_synced: ISO timestamp of last sync

    Returns:
        Result containing modified and deleted entity IDs
    """
    if not last_synced:
        raise ValueError("last_synced parameter is required")

    modified_entity_ids = []
    deleted_entity_ids = []

    try:
        # Get all entities using the entities endpoint
        # This is more efficient than checking each entity individually
        page = 1
        all_entities = {}

        while page <= 20:  # Reasonable limit to avoid infinite loops
            batch = self.service.client.entities(page=page, limit=100)
            if not batch:
                break

            for entity_data in batch:
                entity_id = entity_data.get("id")
                if entity_id:
                    all_entities[entity_id] = entity_data

            if len(batch) < 100:
                break
            page += 1

        # Check each requested entity
        for entity_id in entity_ids:
            if entity_id in all_entities:
                entity_data = all_entities[entity_id]
                updated_at = entity_data.get("updated_at")

                if updated_at and updated_at > last_synced:
                    modified_entity_ids.append(entity_id)
            else:
                # Entity not found - might be deleted
                deleted_entity_ids.append(entity_id)

        # Get current timestamp
        check_timestamp = datetime.now(timezone.utc).isoformat()

        return {
            "modified_entity_ids": modified_entity_ids,
            "deleted_entity_ids": deleted_entity_ids,
            "check_timestamp": check_timestamp,
        }

    except Exception as e:
        logger.error(f"Check entity updates failed: {e}")
        raise

src/mcp_kanka/tools.py:170-190 (handler)

The MCP tool handler that receives parameters, validates that last_synced is provided, and delegates to the operations layer. This is the entry point called by the MCP server.

async def handle_check_entity_updates(**params: Any) -> CheckEntityUpdatesResult:
    """
    Check which entity_ids have been modified since last sync.

    Args:
        **params: Parameters from CheckEntityUpdatesParams

    Returns:
        Check result with modified and deleted entity IDs
    """
    entity_ids = params.get("entity_ids", [])
    last_synced = params.get("last_synced")

    # Validate last_synced is provided
    if not last_synced:
        raise ValueError("last_synced parameter is required")

    operations = get_operations()

    # Delegate to operations layer
    return await operations.check_entity_updates(entity_ids, last_synced)

src/mcp_kanka/types.py:314-326 (schema)

Type definitions for CheckEntityUpdatesParams (entity_ids list and last_synced timestamp) and CheckEntityUpdatesResult (modified_entity_ids, deleted_entity_ids, check_timestamp).

class CheckEntityUpdatesParams(TypedDict):
    """Parameters for check_entity_updates tool."""

    entity_ids: list[int]
    last_synced: str  # ISO 8601 timestamp


class CheckEntityUpdatesResult(TypedDict):
    """Result of checking entity updates."""

    modified_entity_ids: list[int]
    deleted_entity_ids: list[int]  # If API provides this
    check_timestamp: str  # ISO 8601 timestamp

src/mcp_kanka/__main__.py:369-386 (registration)

Tool registration with the MCP server, including the tool name, description, and JSON Schema for input validation (entity_ids array of integers and last_synced string).

    name="check_entity_updates",
    description="Check which entity_ids have been modified since last sync",
    inputSchema={
        "type": "object",
        "properties": {
            "entity_ids": {
                "type": "array",
                "items": {"type": "integer"},
                "description": "Array of entity IDs to check",
            },
            "last_synced": {
                "type": "string",
                "description": "ISO 8601 timestamp to check updates since",
            },
        },
        "required": ["entity_ids", "last_synced"],
    },
),

src/mcp_kanka/__main__.py:413-414 (registration)
Tool dispatch logic that routes check_entity_updates calls to the handle_check_entity_updates function.
```
elif name == "check_entity_updates":
    result = await handle_check_entity_updates(**arguments)
```

MCP-Kanka