list_posts
Retrieve all posts for a specific Beehiiv publication using the publication ID. This tool integrates with the Beehiiv API v2 to provide access to post data for analysis or management purposes.
Instructions
List all posts for a given publication.
Args:
publication_id: e.g. 'pub_00000000-0000-0000-0000-000000000000'Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| publication_id | Yes |
Implementation Reference
- beehiiv_server.py:62-80 (handler)The handler function for the 'list_posts' tool. It uses the beehiiv_request helper to fetch the latest 5 confirmed posts from the specified publication and formats them as a string list of ID: title.
@mcp.tool() async def list_posts(publication_id: str) -> str: """ List all posts for a given publication. Args: publication_id: e.g. 'pub_00000000-0000-0000-0000-000000000000' """ params = { "order_by": "publish_date", "direction": "desc", "limit": 5, "status": "confirmed" } path = f"/publications/{publication_id}/posts" data = await beehiiv_request("GET", path, params=params) if not data or "data" not in data: return f"API error: {data.get('error', 'Unknown error')}" return "\n".join(f"{p['id']}: {p['title']}" for p in data["data"]) - beehiiv_server.py:17-50 (helper)Shared helper function that performs authenticated HTTP requests to the Beehiiv API v2, handling errors and used by the list_posts handler.
async def beehiiv_request( method: str, path: str, params: Optional[dict[str, Any]] = None, json_body: Optional[dict[str, Any]] = None ) -> Optional[dict[str, Any]]: """ Helper to call the beehiiv API v2. Args: method: HTTP method (GET, POST, etc.) path: API path (e.g. '/publications') params: Query parameters json_body: Request JSON body """ headers = { "Authorization": f"Bearer {BEEHIIV_API_KEY}", "Content-Type": "application/json" } url = f"{BASE_URL}{path}" async with httpx.AsyncClient() as client: try: response = await client.request( method, url, headers=headers, params=params, json=json_body, timeout=30.0 ) response.raise_for_status() return response.json() except httpx.HTTPError as e: return {"error": str(e)}