list_posts

Instructions

Implementation Reference

• src/mcp_server.py:156-216 (handler)

  MCP tool handler and registration for 'list_posts'. Validates inputs, fetches posts from ChromaService, resolves URLs, serializes, and returns paginated results.

  @mcp.tool()
  async def list_posts(
      sort_by: str = "newest",
      page: int = 1,
      limit: int = 10,
  ) -> dict[str, Any]:
      """
      List blog posts with pagination.
  
      Args:
          sort_by: Sort order - 'newest' or 'oldest' (default: 'newest')
          page: Page number (default: 1)
          limit: Number of posts per page, max 10 (default: 10)
  
      Returns:
          List of post summaries with metadata
      """
      if limit > settings.max_posts_per_page:
          limit = settings.max_posts_per_page
  
      if page < 1:
          page = 1
  
      offset = (page - 1) * limit
  
      chroma_service = await get_chroma_service()
      posts = await chroma_service.list_posts(
          limit=limit,
          offset=offset,
          sort_by=sort_by,
      )
  
      serialized_posts: list[dict[str, Any]] = []
      for post in posts:
          resolved_url = _canonical_post_url(post)
          if not resolved_url:
              logger.debug("Skipping post without canonical URL: %s", post.id)
              continue
  
          serialized_posts.append(
              {
                  "id": resolved_url,
                  "title": post.title,
                  "excerpt": post.excerpt,
                  "url": resolved_url,
                  "published_at": post.published_at.isoformat() if post.published_at else None,
                  "updated_at": post.updated_at.isoformat() if post.updated_at else None,
                  "tags": post.tags,
                  "authors": post.authors,
              }
          )
  
      return {
          "posts": serialized_posts,
          "pagination": {
              "page": page,
              "limit": limit,
              "sort_by": sort_by,
          },
      }

• src/chroma_service.py:197-257 (helper)

  Helper method in ChromaService that retrieves all post chunks from the database, deduplicates by post slug, sorts by date, applies pagination, and returns PostSummary objects.

  async def list_posts(
      self,
      limit: int = 10,
      offset: int = 0,
      sort_by: str = "newest",
  ) -> list[PostSummary]:
      # Chroma Cloud has a limit of 300 items per request
      all_results = self.collection.get(limit=300)
  
      if not all_results["ids"]:
          return []
  
      posts_map: dict[str, dict[str, Any]] = {}
  
      for i, metadata in enumerate(all_results["metadatas"] or []):
          slug = str(metadata.get("post_slug", ""))
          if slug and slug not in posts_map:
              posts_map[slug] = {
                  "id": metadata.get("post_id", ""),
                  "slug": slug,
                  "title": metadata.get("post_title", ""),
                  "url": metadata.get("post_url", ""),
                  "excerpt": (
                      all_results["documents"][i][:200] if all_results["documents"] else None
                  ),
                  "published_at": metadata.get("published_at"),
                  "updated_at": metadata.get("updated_at"),
                  "tags": str(metadata.get("tags", "")).split(",")
                  if metadata.get("tags")
                  else [],
                  "authors": str(metadata.get("authors", "")).split(",")
                  if metadata.get("authors")
                  else [],
              }
  
      posts = list(posts_map.values())
  
      if sort_by == "newest":
          posts.sort(key=lambda x: x.get("published_at", ""), reverse=True)
      elif sort_by == "oldest":
          posts.sort(key=lambda x: x.get("published_at", ""))
  
      paginated_posts = posts[offset : offset + limit]
  
      return [
          PostSummary(
              id=p["id"],
              slug=p["slug"],
              title=p["title"],
              excerpt=p["excerpt"],
              url=p["url"],
              published_at=datetime.fromisoformat(p["published_at"])
              if p.get("published_at")
              else None,
              updated_at=datetime.fromisoformat(p["updated_at"]) if p.get("updated_at") else None,
              tags=p["tags"],
              authors=p["authors"],
          )
          for p in paginated_posts
      ]