MCP-Kanka

Instructions

Implementation Reference

• src/mcp_kanka/tools.py:84-99 (handler)

  Main handler function for get_entities tool that receives MCP tool calls and delegates to operations layer

  async def handle_get_entities(**params: Any) -> list[GetEntityResult]:
      """
      Retrieve specific entities by ID.
  
      Args:
          **params: Parameters from GetEntitiesParams
  
      Returns:
          List of entity results
      """
      entity_ids = params.get("entity_ids", [])
      include_posts = params.get("include_posts", False)
      operations = get_operations()
  
      # Delegate to operations layer
      return await operations.get_entities(entity_ids, include_posts)

• src/mcp_kanka/types.py:77-81 (schema)

  Input parameter schema defining the structure of parameters accepted by get_entities tool

  class GetEntitiesParams(TypedDict):
      """Parameters for get_entities tool."""
  
      entity_ids: list[int]
      include_posts: bool | None

• src/mcp_kanka/__main__.py:236-254 (registration)

  Tool registration defining the MCP tool metadata including name, description, and input schema

      name="get_entities",
      description="Retrieve specific entities by ID with their posts",
      inputSchema={
          "type": "object",
          "properties": {
              "entity_ids": {
                  "type": "array",
                  "items": {"type": "integer"},
                  "description": "Array of entity IDs to retrieve",
              },
              "include_posts": {
                  "type": "boolean",
                  "description": "Include posts for each entity",
                  "default": False,
              },
          },
          "required": ["entity_ids"],
      },
  ),

• src/mcp_kanka/operations.py:464-530 (handler)

  Core business logic implementation that fetches entities from Kanka service and formats results with error handling

  async def get_entities(
      self, entity_ids: list[int], include_posts: bool = False
  ) -> list[GetEntityResult]:
      """Get specific entities by ID.
  
      Args:
          entity_ids: List of entity IDs to retrieve
          include_posts: Whether to include posts for each entity
  
      Returns:
          List of results, one per entity
      """
      results = []
      for entity_id in entity_ids:
          try:
              # Get entity
              entity = self.service.get_entity_by_id(entity_id, include_posts)
  
              if entity:
                  result: GetEntityResult = {
                      "id": entity["id"],
                      "entity_id": entity["entity_id"],
                      "name": entity["name"],
                      "entity_type": entity["entity_type"],
                      "type": entity.get("type"),
                      "entry": entity.get("entry"),
                      "tags": entity.get("tags", []),
                      "is_hidden": entity.get("is_hidden", False),
                      "created_at": entity.get("created_at"),
                      "updated_at": entity.get("updated_at"),
                      "success": True,
                      "error": None,
                  }
  
                  # Add quest-specific fields
                  if entity.get("entity_type") == "quest":
                      result["is_completed"] = entity.get("is_completed")
  
                  # Add all image fields (they should always be present from service layer)
                  result["image"] = entity.get("image")
                  result["image_full"] = entity.get("image_full")
                  result["image_thumb"] = entity.get("image_thumb")
                  result["image_uuid"] = entity.get("image_uuid")
                  result["header_uuid"] = entity.get("header_uuid")
  
                  if include_posts:
                      result["posts"] = entity.get("posts", [])
  
                  results.append(result)
              else:
                  not_found_result: GetEntityResult = {
                      "entity_id": entity_id,
                      "success": False,
                      "error": f"Entity {entity_id} not found",
                  }
                  results.append(not_found_result)
  
          except Exception as e:
              logger.error(f"Failed to get entity {entity_id}: {e}")
              error_result: GetEntityResult = {
                  "entity_id": entity_id,
                  "success": False,
                  "error": str(e),
              }
              results.append(error_result)
  
      return results

• src/mcp_kanka/types.py:216-237 (schema)

  Output result schema defining the structure of responses returned by get_entities tool

  class GetEntityResult(TypedDict, total=False):
      """Result of getting an entity."""
  
      id: int | None
      entity_id: int
      name: str | None
      entity_type: EntityType | None
      type: str | None
      entry: str | None
      tags: list[str] | None
      is_hidden: bool | None
      created_at: str | None  # ISO 8601 timestamp
      updated_at: str | None  # ISO 8601 timestamp
      posts: list[PostData] | None
      success: bool
      error: str | None
      is_completed: bool | None  # For quests only
      image: str | None  # Local path to the picture
      image_full: str | None  # URL to the full picture
      image_thumb: str | None  # URL to the thumbnail
      image_uuid: str | None  # Image gallery UUID
      header_uuid: str | None  # Header image gallery UUID