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",
              },
          },
      },
  ),