MCP-Kanka

Instructions

Implementation Reference

• src/mcp_kanka/operations.py:87-299 (handler)

  Core implementation of find_entities operation in KankaOperations class. Handles entity retrieval with search and filtering capabilities, validates entity types, applies client-side filters (name, type, tags, date range), handles pagination, and returns results with sync metadata.
  async def find_entities( self, query: str | None = None, entity_type: str | None = None, name: str | None = None, name_exact: bool = False, name_fuzzy: bool = False, type: str | None = None, tags: list[str] | None = None, date_range: dict[str, str] | None = None, include_full: bool = True, page: int = 1, limit: int = 25, last_synced: str | None = None, ) -> dict[str, Any]: """Find entities with search and filtering capabilities. Args: query: Search term for full-text search across names and content entity_type: Type of entity to search for name: Filter by entity name name_exact: Use exact name matching (case-insensitive) name_fuzzy: Use fuzzy name matching (typo-tolerant) type: Filter by custom type field tags: Filter by tags (must have all specified tags) date_range: Date range filter for journals include_full: Whether to include full entity details page: Page number for pagination limit: Number of results per page (0 for all) last_synced: ISO timestamp to get only entities modified after this time Returns: Dictionary with entities and sync_info """ # Validate entity type if provided valid_types = [ "character", "creature", "location", "organization", "race", "note", "journal", "quest", ] if entity_type and entity_type not in valid_types: logger.error( f"Invalid entity_type: {entity_type}. Must be one of: {', '.join(valid_types)}" ) return {"entities": [], "sync_info": {}} try: # Step 1: Get entities if query: # For content search, we need full entities entities = [] if entity_type: # Search specific entity type # Cast to EntityType since we validated it above from typing import cast from .types import EntityType entity_objects = self.service.list_entities( cast(EntityType, entity_type), page=1, limit=0, last_sync=last_synced, related=include_full, ) for obj in entity_objects: entity_dict = self.service._entity_to_dict(obj, entity_type) entities.append(entity_dict) else: # Search across all entity types from .types import EntityType entity_types: list[EntityType] = [ "character", "creature", "location", "organization", "race", "note", "journal", "quest", ] for et in entity_types: try: entity_objects = self.service.list_entities( et, page=1, limit=0, last_sync=last_synced, related=include_full, ) for obj in entity_objects: entity_dict = self.service._entity_to_dict(obj, et) entities.append(entity_dict) except Exception as e: logger.debug(f"Could not search {et}: {e}") continue # Apply content search entities = search_in_content(entities, query) # If not including full details, strip to minimal data if not include_full: minimal_entities = [] for entity in entities: minimal_entities.append( { "entity_id": entity["entity_id"], "name": entity["name"], "entity_type": entity["entity_type"], } ) entities = minimal_entities else: # List entities of specific type (no search) if not entity_type: # No entity type specified, can't list all return {"entities": [], "sync_info": {}} # Get all entities of this type # Cast to EntityType since we validated it above from typing import cast from .types import EntityType entity_objects = self.service.list_entities( cast(EntityType, entity_type), page=1, limit=0, last_sync=last_synced, related=include_full, ) # Convert to dictionaries entities = [] for obj in entity_objects: entity_dict = self.service._entity_to_dict(obj, entity_type) entities.append(entity_dict) # Step 2: Apply client-side filters if name: entities = filter_entities_by_name( entities, name, exact=name_exact, fuzzy=name_fuzzy ) if type: entities = filter_entities_by_type(entities, type) if tags: entities = filter_entities_by_tags(entities, tags) if date_range and entity_type == "journal": start = date_range.get("start") end = date_range.get("end") if start and end: entities = filter_journals_by_date_range(entities, start, end) # Don't apply content search if we already used the search API # The search API already searched content # Step 3: Paginate results paginated, total_pages, total_items = paginate_results( entities, page, limit ) # Step 4: Calculate sync metadata # Find newest updated_at timestamp newest_updated_at = None for entity in paginated: if entity.get("updated_at") and ( newest_updated_at is None or entity["updated_at"] > newest_updated_at ): newest_updated_at = entity["updated_at"] # Build sync info sync_info = { "request_timestamp": datetime.now(timezone.utc).isoformat(), "newest_updated_at": newest_updated_at, "total_count": total_items, "returned_count": len(paginated), } # Step 5: Format results based on include_full if not include_full: # Return minimal data formatted_entities = [ { "entity_id": e["entity_id"], "name": e["name"], "entity_type": e["entity_type"], } for e in paginated ] else: # Return full data formatted_entities = paginated # Return the new response structure return { "entities": formatted_entities, "sync_info": sync_info, } except Exception as e: logger.error(f"find_entities failed: {e}") raise
• src/mcp_kanka/tools.py:21-47 (handler)

  MCP tool handler for find_entities. Receives parameters from MCP client and delegates to the operations layer. Extracts query, entity_type, name, tags, date_range, include_full, page, limit, and last_synced parameters.
  async def handle_find_entities(**params: Any) -> dict[str, Any]: """ Find entities by search and/or filtering. Args: **params: Parameters from FindEntitiesParams Returns: Dictionary with entities and sync_info """ operations = get_operations() # Delegate to operations layer return await operations.find_entities( query=params.get("query"), entity_type=params.get("entity_type"), name=params.get("name"), name_exact=params.get("name_exact", False), name_fuzzy=params.get("name_fuzzy", False), type=params.get("type"), tags=params.get("tags", []), date_range=params.get("date_range"), include_full=params.get("include_full", True), page=params.get("page", 1), limit=params.get("limit", 25), last_synced=params.get("last_synced"), )
• src/mcp_kanka/types.py:26-40 (schema)

  FindEntitiesParams TypedDict defining input parameters for the find_entities tool including query, entity_type, name filters, type, tags, date_range, include_full, pagination options, and last_synced timestamp.
  class FindEntitiesParams(TypedDict, total=False): """Parameters for find_entities tool.""" query: str | None entity_type: EntityType | None name: str | None name_exact: bool | None name_fuzzy: bool | None type: str | None tags: list[str] | None date_range: DateRange | None include_full: bool | None page: int | None limit: int | None last_synced: str | None # ISO 8601 timestamp
• src/mcp_kanka/types.py:190-194 (schema)

  FindEntitiesResponse TypedDict defining output structure with entities list (EntityMinimal or EntityFull) and sync_info containing request_timestamp, newest_updated_at, total_count, and returned_count.
  class FindEntitiesResponse(TypedDict): """Response structure for find_entities with sync metadata.""" entities: list[EntityMinimal | EntityFull] sync_info: SyncInfo
• src/mcp_kanka/__main__.py:72-147 (registration)

  Tool registration for find_entities with complete input schema defining all parameters, their types, descriptions, and defaults. Maps tool name to handle_find_entities function at line 398.
  name="find_entities", description="Find entities by search and/or filtering", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Search term (searches names and content)", }, "entity_type": { "type": "string", "enum": [ "character", "creature", "location", "organization", "race", "note", "journal", "quest", ], "description": "Entity type to filter by", }, "name": { "type": "string", "description": "Filter by name (partial match by default, e.g. 'Test' matches 'Test Character')", }, "name_exact": { "type": "boolean", "description": "Use exact matching on name filter (case-insensitive)", "default": False, }, "name_fuzzy": { "type": "boolean", "description": "Use fuzzy matching on name filter (typo-tolerant)", "default": False, }, "type": { "type": "string", "description": "Filter by Type field (e.g., 'NPC', 'City')", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "Filter by tags (matches entities having ALL specified tags)", }, "date_range": { "type": "object", "properties": { "start": {"type": "string", "format": "date"}, "end": {"type": "string", "format": "date"}, }, "description": "For filtering journals by date", }, "include_full": { "type": "boolean", "description": "Include full entity details", "default": True, }, "page": { "type": "integer", "description": "Page number for pagination", "default": 1, }, "limit": { "type": "integer", "description": "Results per page (default 25, max 100, use 0 for all)", "default": 25, }, "last_synced": { "type": "string", "description": "ISO 8601 timestamp to get only entities modified after this time", }, }, }, ),