MCP-Kanka

Instructions

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)