diff --git a/README.md b/README.md index 5151f8a..64a4a0b 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,8 @@ pip install late-sdk ## Quick Start ```python -from late import Late +from datetime import datetime, timedelta +from late import Late, Platform client = Late(api_key="your_api_key") @@ -25,11 +26,9 @@ client = Late(api_key="your_api_key") accounts = client.accounts.list() # Create a scheduled post -from datetime import datetime, timedelta - post = client.posts.create( content="Hello from Late!", - platforms=[{"platform": "twitter", "accountId": "your_account_id"}], + platforms=[{"platform": Platform.TWITTER, "accountId": "your_account_id"}], scheduled_for=datetime.now() + timedelta(hours=1), ) ``` @@ -119,13 +118,13 @@ Since Claude can't access local files, use the browser upload flow: | Command | What it does | |---------|--------------| -| `list_accounts` | Show connected social accounts | -| `create_post` | Create scheduled or immediate post | -| `publish_now` | Publish immediately | -| `cross_post` | Post to multiple platforms | -| `list_posts` | Show your posts | -| `retry_post` | Retry a failed post | -| `generate_upload_link` | Get link to upload media | +| `accounts_list` | Show connected social accounts | +| `posts_create` | Create scheduled, immediate, or draft post | +| `posts_publish_now` | Publish immediately | +| `posts_cross_post` | Post to multiple platforms | +| `posts_list` | Show your posts | +| `posts_retry` | Retry a failed post | +| `media_generate_upload_link` | Get link to upload media | --- @@ -144,22 +143,27 @@ async def main(): asyncio.run(main()) ``` -### AI Content Generation +### AI Content Generation (Experimental) ```bash pip install late-sdk[ai] ``` ```python +from late import Platform, CaptionTone from late.ai import ContentGenerator, GenerateRequest -generator = ContentGenerator(provider="openai", api_key="sk-...") +generator = ContentGenerator( + provider="openai", + api_key="sk-...", + model="gpt-4o-mini", # or gpt-4o, gpt-4-turbo, etc. +) response = generator.generate( GenerateRequest( prompt="Write a tweet about Python", - platform="twitter", - tone="casual", + platform=Platform.TWITTER, + tone=CaptionTone.CASUAL, ) ) @@ -185,6 +189,7 @@ results = pipeline.schedule("posts.csv") ### Cross-Posting ```python +from late import Platform from late.pipelines import CrossPosterPipeline, PlatformConfig cross_poster = CrossPosterPipeline(client) @@ -192,8 +197,8 @@ cross_poster = CrossPosterPipeline(client) results = await cross_poster.post( content="Big announcement!", platforms=[ - PlatformConfig("twitter", "tw_123"), - PlatformConfig("linkedin", "li_456", delay_minutes=5), + PlatformConfig(Platform.TWITTER, "tw_123"), + PlatformConfig(Platform.LINKEDIN, "li_456", delay_minutes=5), ], ) ``` diff --git a/pyproject.toml b/pyproject.toml index 1fb35c6..8e7b443 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "late-sdk" -version = "1.1.0" +version = "1.1.1" description = "Python SDK for Late API - Social Media Scheduling" readme = "README.md" requires-python = ">=3.10" @@ -113,6 +113,10 @@ ignore = [ "B008", # do not perform function calls in argument defaults ] +[tool.ruff.lint.per-file-ignores] +"**/models/_generated/*" = ["UP006", "UP007", "UP035", "W291"] # Allow old-style annotations and trailing whitespace in generated code +"**/models/_generated/**" = ["UP006", "UP007", "UP035", "W291"] + [tool.ruff.lint.isort] known-first-party = ["late"] diff --git a/scripts/generate_mcp_docs.py b/scripts/generate_mcp_docs.py index a5d19d8..2dd3f76 100644 --- a/scripts/generate_mcp_docs.py +++ b/scripts/generate_mcp_docs.py @@ -7,6 +7,8 @@ This script generates MDX documentation from the centralized tool definitions in src/late/mcp/tool_definitions.py + +The generated output can be copied into the docs site. """ import sys @@ -15,18 +17,19 @@ # Add src to path for imports sys.path.insert(0, str(Path(__file__).parent.parent / "src")) -from late.mcp.tool_definitions import generate_mdx_docs, TOOL_DEFINITIONS +from late.mcp.tool_definitions import TOOL_DEFINITIONS, generate_mdx_tools_reference -def main(): +def main() -> None: """Generate and print MDX documentation.""" print("=" * 60) print("MCP Tool Documentation (generated from tool_definitions.py)") print("=" * 60) print() - print(generate_mdx_docs()) + print(generate_mdx_tools_reference()) print() print("=" * 60) + print(f"Total tools: {len(TOOL_DEFINITIONS)}") print("Copy the above into claude-mcp.mdx under '## Tool Reference'") print("=" * 60) diff --git a/src/late/__init__.py b/src/late/__init__.py index f9fa4cf..06b5506 100644 --- a/src/late/__init__.py +++ b/src/late/__init__.py @@ -31,7 +31,7 @@ Visibility, ) -__version__ = "1.1.0" +__version__ = "1.1.1" __all__ = [ # Client diff --git a/src/late/mcp/server.py b/src/late/mcp/server.py index 00d5958..47956c5 100644 --- a/src/late/mcp/server.py +++ b/src/late/mcp/server.py @@ -25,11 +25,14 @@ import os from datetime import datetime, timedelta +from typing import Any from mcp.server.fastmcp import FastMCP from late import Late, MediaType, PostStatus +from .tool_definitions import use_tool_def + # Initialize MCP server mcp = FastMCP( "Late", @@ -61,47 +64,38 @@ def _get_client() -> Late: @mcp.tool() +@use_tool_def("accounts_list") def accounts_list() -> str: - """ - List all connected social media accounts. - - Returns the platform, username, and account ID for each connected account. - """ client = _get_client() response = client.accounts.list() - accounts = response.get("accounts", []) + accounts = response.accounts or [] if not accounts: return "No accounts connected. Connect accounts at https://getlate.dev" lines = [f"Found {len(accounts)} connected account(s):\n"] for acc in accounts: - username = acc.get("username") or acc.get("name") or acc["_id"] - lines.append(f"- {acc['platform']}: {username} (ID: {acc['_id']})") + username = acc.username or acc.displayName or acc.field_id + lines.append(f"- {acc.platform}: {username} (ID: {acc.field_id})") return "\n".join(lines) @mcp.tool() +@use_tool_def("accounts_get") def accounts_get(platform: str) -> str: - """ - Get account details for a specific platform. - - Args: - platform: Platform name (twitter, instagram, linkedin, tiktok, bluesky, facebook, youtube, pinterest, threads) - """ client = _get_client() response = client.accounts.list() - accounts = response.get("accounts", []) + accounts = response.accounts or [] - matching = [a for a in accounts if a["platform"].lower() == platform.lower()] + matching = [a for a in accounts if a.platform and a.platform.lower() == platform.lower()] if not matching: - available = list({a["platform"] for a in accounts}) + available = list({a.platform for a in accounts if a.platform}) return f"No {platform} account found. Available: {', '.join(available)}" acc = matching[0] - return f"Platform: {acc['platform']}\nUsername: {acc.get('username', 'N/A')}\nID: {acc['_id']}" + return f"Platform: {acc.platform}\nUsername: {acc.username or 'N/A'}\nID: {acc.field_id}" # ============================================================================ @@ -110,80 +104,67 @@ def accounts_get(platform: str) -> str: @mcp.tool() +@use_tool_def("profiles_list") def profiles_list() -> str: - """ - List all profiles. - - Profiles group multiple social accounts together for easier management. - """ client = _get_client() response = client.profiles.list() - profiles = response.get("profiles", []) + profiles = response.profiles or [] if not profiles: return "No profiles found." lines = [f"Found {len(profiles)} profile(s):\n"] for profile in profiles: - default = " (default)" if profile.get("isDefault") else "" - color = f" [{profile.get('color', '')}]" if profile.get("color") else "" - lines.append(f"- {profile['name']}{default}{color} (ID: {profile['_id']})") - if profile.get("description"): - lines.append(f" Description: {profile['description']}") + default = " (default)" if profile.isDefault else "" + color = f" [{profile.color}]" if profile.color else "" + lines.append(f"- {profile.name}{default}{color} (ID: {profile.field_id})") + if profile.description: + lines.append(f" Description: {profile.description}") return "\n".join(lines) @mcp.tool() +@use_tool_def("profiles_get") def profiles_get(profile_id: str) -> str: - """ - Get details of a specific profile. - - Args: - profile_id: The profile ID - """ client = _get_client() response = client.profiles.get(profile_id) - profile = response.get("profile", response) + profile = response.profile + if not profile: + return f"Profile {profile_id} not found." lines = [ - f"Name: {profile['name']}", - f"ID: {profile['_id']}", - f"Default: {'Yes' if profile.get('isDefault') else 'No'}", + f"Name: {profile.name}", + f"ID: {profile.field_id}", + f"Default: {'Yes' if profile.isDefault else 'No'}", ] - if profile.get("description"): - lines.append(f"Description: {profile['description']}") - if profile.get("color"): - lines.append(f"Color: {profile['color']}") + if profile.description: + lines.append(f"Description: {profile.description}") + if profile.color: + lines.append(f"Color: {profile.color}") return "\n".join(lines) @mcp.tool() +@use_tool_def("profiles_create") def profiles_create(name: str, description: str = "", color: str = "") -> str: - """ - Create a new profile. - - Args: - name: Profile name (required) - description: Optional description - color: Optional hex color (e.g., '#4CAF50') - """ client = _get_client() - params = {"name": name} + params: dict[str, str] = {"name": name} if description: params["description"] = description if color: params["color"] = color response = client.profiles.create(**params) - profile = response.get("profile", {}) + profile = response.profile - return f"✅ Profile created!\nName: {profile.get('name')}\nID: {profile.get('_id')}" + return f"✅ Profile created!\nName: {profile.name if profile else 'N/A'}\nID: {profile.field_id if profile else 'N/A'}" @mcp.tool() +@use_tool_def("profiles_update") def profiles_update( profile_id: str, name: str = "", @@ -191,19 +172,9 @@ def profiles_update( color: str = "", is_default: bool = False, ) -> str: - """ - Update an existing profile. - - Args: - profile_id: The profile ID to update - name: New name (leave empty to keep current) - description: New description (leave empty to keep current) - color: New hex color (leave empty to keep current) - is_default: Set as default profile - """ client = _get_client() - params = {} + params: dict[str, str | bool] = {} if name: params["name"] = name if description: @@ -217,21 +188,14 @@ def profiles_update( return "⚠️ No changes specified. Provide at least one field to update." response = client.profiles.update(profile_id, **params) - profile = response.get("profile", {}) + profile = response.profile - return f"✅ Profile updated!\nName: {profile.get('name')}\nID: {profile.get('_id')}" + return f"✅ Profile updated!\nName: {profile.name if profile else 'N/A'}\nID: {profile.field_id if profile else 'N/A'}" @mcp.tool() +@use_tool_def("profiles_delete") def profiles_delete(profile_id: str) -> str: - """ - Delete a profile. - - Note: Profile must have no connected accounts. - - Args: - profile_id: The profile ID to delete - """ client = _get_client() client.profiles.delete(profile_id) return f"✅ Profile {profile_id} deleted" @@ -243,76 +207,68 @@ def profiles_delete(profile_id: str) -> str: @mcp.tool() +@use_tool_def("posts_list") def posts_list(status: str = "", limit: int = 10) -> str: - """ - List posts with optional filtering. - - Args: - status: Filter by status (scheduled, published, failed, draft). Empty for all. - limit: Maximum number of posts to return (default 10) - """ client = _get_client() - params = {"limit": limit} + params: dict[str, str | int] = {"limit": limit} if status: params["status"] = status response = client.posts.list(**params) - posts = response.get("posts", []) + posts = response.posts or [] if not posts: return f"No posts found{f' with status {status}' if status else ''}." lines = [f"Found {len(posts)} post(s):\n"] for post in posts: - content_preview = ( - post["content"][:60] + "..." - if len(post["content"]) > 60 - else post["content"] + content = post.content or "" + content_preview = content[:60] + "..." if len(content) > 60 else content + platforms = ", ".join( + t.platform or "?" for t in (post.platforms or []) ) - platforms = ", ".join(t.get("platform", "?") for t in post.get("platforms", [])) - lines.append(f"- [{post['status']}] {content_preview}") - lines.append(f" Platforms: {platforms} | ID: {post['_id']}") + status = post.status.value if post.status else "unknown" + lines.append(f"- [{status}] {content_preview}") + lines.append(f" Platforms: {platforms} | ID: {post.field_id}") return "\n".join(lines) @mcp.tool() +@use_tool_def("posts_get") def posts_get(post_id: str) -> str: - """ - Get details of a specific post by ID. - - Args: - post_id: The post ID to retrieve - """ client = _get_client() response = client.posts.get(post_id) - post = response.get("post", response) + post = response.post + if not post: + return f"Post {post_id} not found." - content_preview = ( - post["content"][:100] + "..." if len(post["content"]) > 100 else post["content"] - ) - platforms = ", ".join(t.get("platform", "?") for t in post.get("platforms", [])) + content = post.content or "" + content_preview = content[:100] + "..." if len(content) > 100 else content + platforms = ", ".join(t.platform or "?" for t in (post.platforms or [])) + status = post.status.value if post.status else "unknown" lines = [ - f"Post ID: {post['_id']}", - f"Status: {post['status']}", + f"Post ID: {post.field_id}", + f"Status: {status}", f"Platforms: {platforms}", f"Content: {content_preview}", ] - if post.get("scheduledFor"): - lines.append(f"Scheduled for: {post['scheduledFor']}") + if post.scheduledFor: + lines.append(f"Scheduled for: {post.scheduledFor}") - if post.get("publishedAt"): - lines.append(f"Published at: {post['publishedAt']}") + if hasattr(post, "publishedAt") and post.publishedAt: + lines.append(f"Published at: {post.publishedAt}") - if post.get("error"): - lines.append(f"Error: {post['error']}") + if post.metadata and post.metadata.get("error"): + lines.append(f"Error: {post.metadata['error']}") return "\n".join(lines) @mcp.tool() +@use_tool_def("posts_create") def posts_create( content: str, platform: str, @@ -322,42 +278,26 @@ def posts_create( media_urls: str = "", title: str = "", ) -> str: - """ - Create a new social media post, optionally with media. - - Scheduling behavior: - - is_draft=True: Save as draft (no scheduling, can edit later) - - publish_now=True: Publish immediately - - Neither: Schedule for schedule_minutes from now (default: 60 min) - - Args: - content: The post content/text - platform: Target platform (twitter, instagram, linkedin, tiktok, bluesky, facebook, youtube, pinterest, threads) - is_draft: Save as draft without scheduling. Draft posts can be edited and scheduled later (default: False) - publish_now: Publish immediately instead of scheduling (default: False) - schedule_minutes: Minutes from now to schedule (ignored if publish_now=True or is_draft=True). Default 60 min. - media_urls: Comma-separated URLs of media files to attach. Optional. - title: Optional title (required for YouTube, recommended for Pinterest) - """ client = _get_client() # Find account for platform - accounts = client.accounts.list().get("accounts", []) - matching = [a for a in accounts if a["platform"].lower() == platform.lower()] + accounts_response = client.accounts.list() + accounts = accounts_response.accounts or [] + matching = [a for a in accounts if a.platform and a.platform.lower() == platform.lower()] if not matching: - available = list({a["platform"] for a in accounts}) + available = list({a.platform for a in accounts if a.platform}) return f"No {platform} account connected. Available platforms: {', '.join(available)}" account = matching[0] # Build request - params = { + params: dict[str, Any] = { "content": content, "platforms": [ { - "platform": account["platform"], - "accountId": account["_id"], + "platform": account.platform, + "accountId": account.field_id, } ], } @@ -389,40 +329,35 @@ def posts_create( params["scheduled_for"] = datetime.now() + timedelta(minutes=minutes) response = client.posts.create(**params) - post = response.get("post", {}) + post = response.post - username = account.get("username") or account.get("name") or account["_id"] + username = account.username or account.displayName or account.field_id media_info = ( f" with {len(params.get('media_items', []))} media file(s)" if params.get("media_items") else "" ) + post_id = post.field_id if post else "N/A" if is_draft: - return f"📝 Draft saved for {platform} (@{username}){media_info}\nPost ID: {post.get('_id', 'N/A')}\nStatus: draft" + return f"📝 Draft saved for {platform} (@{username}){media_info}\nPost ID: {post_id}\nStatus: draft" elif publish_now: - return f"✅ Published to {platform} (@{username}){media_info}\nPost ID: {post.get('_id', 'N/A')}" + return f"✅ Published to {platform} (@{username}){media_info}\nPost ID: {post_id}" else: scheduled = params["scheduled_for"].strftime("%Y-%m-%d %H:%M") - return f"✅ Scheduled for {platform} (@{username}){media_info}\nPost ID: {post.get('_id', 'N/A')}\nScheduled: {scheduled}" + return f"✅ Scheduled for {platform} (@{username}){media_info}\nPost ID: {post_id}\nScheduled: {scheduled}" @mcp.tool() +@use_tool_def("posts_publish_now") def posts_publish_now(content: str, platform: str, media_urls: str = "") -> str: - """ - Publish a post immediately to a platform. - - Args: - content: The post content/text - platform: Target platform (twitter, instagram, linkedin, tiktok, bluesky, etc.) - media_urls: Comma-separated URLs of media files to attach. Optional. - """ return posts_create( content=content, platform=platform, publish_now=True, media_urls=media_urls ) @mcp.tool() +@use_tool_def("posts_cross_post") def posts_cross_post( content: str, platforms: str, @@ -430,46 +365,32 @@ def posts_cross_post( publish_now: bool = False, media_urls: str = "", ) -> str: - """ - Post the same content to multiple platforms at once. - - Scheduling behavior: - - is_draft=True: Save as draft (no scheduling, can edit later) - - publish_now=True: Publish immediately - - Neither: Schedule for 1 hour from now - - Args: - content: The post content/text - platforms: Comma-separated list of platforms (e.g., "twitter,linkedin,bluesky") - is_draft: Save as draft without scheduling (default: False) - publish_now: Publish immediately instead of scheduling (default: False) - media_urls: Comma-separated URLs of media files to attach. Optional. - """ client = _get_client() target_platforms = [p.strip().lower() for p in platforms.split(",")] - accounts = client.accounts.list().get("accounts", []) + accounts_response = client.accounts.list() + accounts = accounts_response.accounts or [] platform_targets = [] not_found = [] for platform in target_platforms: - matching = [a for a in accounts if a["platform"].lower() == platform] + matching = [a for a in accounts if a.platform and a.platform.lower() == platform] if matching: platform_targets.append( { - "platform": matching[0]["platform"], - "accountId": matching[0]["_id"], + "platform": matching[0].platform, + "accountId": matching[0].field_id, } ) else: not_found.append(platform) if not platform_targets: - available = list({a["platform"] for a in accounts}) + available = list({a.platform for a in accounts if a.platform}) return f"No matching accounts found. Available: {', '.join(available)}" - params = { + params: dict[str, Any] = { "content": content, "platforms": platform_targets, } @@ -496,7 +417,7 @@ def posts_cross_post( params["scheduled_for"] = datetime.now() + timedelta(hours=1) response = client.posts.create(**params) - post = response.get("post", {}) + post = response.post posted_to = [t["platform"] for t in platform_targets] media_info = ( @@ -505,10 +426,11 @@ def posts_cross_post( else "" ) + post_id = post.field_id if post else "N/A" if is_draft: - result = f"📝 Draft saved for: {', '.join(posted_to)}{media_info}\nPost ID: {post.get('_id', 'N/A')}\nStatus: draft" + result = f"📝 Draft saved for: {', '.join(posted_to)}{media_info}\nPost ID: {post_id}\nStatus: draft" else: - result = f"✅ {'Published' if publish_now else 'Scheduled'} to: {', '.join(posted_to)}{media_info}\nPost ID: {post.get('_id', 'N/A')}" + result = f"✅ {'Published' if publish_now else 'Scheduled'} to: {', '.join(posted_to)}{media_info}\nPost ID: {post_id}" if not_found: result += f"\n⚠️ Accounts not found for: {', '.join(not_found)}" @@ -517,23 +439,13 @@ def posts_cross_post( @mcp.tool() +@use_tool_def("posts_update") def posts_update( post_id: str, content: str = "", scheduled_for: str = "", title: str = "", ) -> str: - """ - Update an existing post. - - Only draft, scheduled, and failed posts can be updated. - - Args: - post_id: The post ID to update - content: New content (leave empty to keep current) - scheduled_for: New schedule time as ISO string (leave empty to keep current) - title: New title (leave empty to keep current) - """ client = _get_client() params = {} @@ -548,41 +460,33 @@ def posts_update( return "⚠️ No changes specified. Provide at least one field to update." response = client.posts.update(post_id, **params) - post = response.get("post", {}) + post = response.post - return f"✅ Post updated!\nID: {post.get('_id')}\nStatus: {post.get('status')}" + post_id_str = post.field_id if post else "N/A" + status = post.status if post else "N/A" + return f"✅ Post updated!\nID: {post_id_str}\nStatus: {status}" @mcp.tool() +@use_tool_def("posts_delete") def posts_delete(post_id: str) -> str: - """ - Delete a post by ID. - - Published posts cannot be deleted. - - Args: - post_id: The post ID to delete - """ client = _get_client() client.posts.delete(post_id) return f"✅ Post {post_id} deleted" @mcp.tool() +@use_tool_def("posts_retry") def posts_retry(post_id: str) -> str: - """ - Retry a failed post. - - Args: - post_id: The ID of the failed post to retry - """ client = _get_client() try: post_response = client.posts.get(post_id) - post = post_response.get("post", post_response) - if post.get("status") != PostStatus.FAILED: - return f"⚠️ Post {post_id} is not in failed status (current: {post.get('status')})" + post = post_response.post + if not post: + return f"❌ Post {post_id} not found" + if post.status != PostStatus.FAILED: + return f"⚠️ Post {post_id} is not in failed status (current: {post.status})" except Exception as e: return f"❌ Could not find post {post_id}: {e}" @@ -594,31 +498,23 @@ def posts_retry(post_id: str) -> str: @mcp.tool() +@use_tool_def("posts_list_failed") def posts_list_failed(limit: int = 10) -> str: - """ - List all failed posts that can be retried. - - Args: - limit: Maximum number of posts to return (default 10) - """ client = _get_client() response = client.posts.list(status=PostStatus.FAILED, limit=limit) - posts = response.get("posts", []) + posts = response.posts or [] if not posts: return "No failed posts found." lines = [f"Found {len(posts)} failed post(s):\n"] for post in posts: - content_preview = ( - post["content"][:50] + "..." - if len(post["content"]) > 50 - else post["content"] - ) - platforms = ", ".join(t.get("platform", "?") for t in post.get("platforms", [])) - error = post.get("error", "Unknown error") + content = post.content or "" + content_preview = content[:50] + "..." if len(content) > 50 else content + platforms = ", ".join(t.platform or "?" for t in (post.platforms or [])) + error = post.metadata.get("error", "Unknown error") if post.metadata else "Unknown error" lines.append(f"- {content_preview}") - lines.append(f" Platforms: {platforms} | ID: {post['_id']}") + lines.append(f" Platforms: {platforms} | ID: {post.field_id}") lines.append(f" Error: {error}") lines.append("") @@ -626,13 +522,11 @@ def posts_list_failed(limit: int = 10) -> str: @mcp.tool() +@use_tool_def("posts_retry_all_failed") def posts_retry_all_failed() -> str: - """ - Retry all failed posts. - """ client = _get_client() response = client.posts.list(status=PostStatus.FAILED, limit=50) - posts = response.get("posts", []) + posts = response.posts or [] if not posts: return "No failed posts to retry." @@ -643,11 +537,11 @@ def posts_retry_all_failed() -> str: for post in posts: try: - client.posts.retry(post["_id"]) + client.posts.retry(post.field_id) success_count += 1 except Exception as e: fail_count += 1 - results.append(f"❌ {post['_id']}: {e}") + results.append(f"❌ {post.field_id}: {e}") summary = f"✅ Retried {success_count} post(s)" if fail_count > 0: @@ -663,29 +557,16 @@ def posts_retry_all_failed() -> str: @mcp.tool() +@use_tool_def("media_generate_upload_link") def media_generate_upload_link() -> str: - """ - Generate a unique upload URL for the user to upload files via browser. - - Use this when the user wants to include images or videos in their post. - The flow is: - 1. Call this tool to get an upload URL - 2. Ask the user to open the URL in their browser - 3. User uploads files through the web interface - 4. Call media_check_upload_status to get the uploaded file URLs - 5. Use those URLs when creating the post with posts_create - - Returns: - Upload URL and token for the user to open in browser - """ client = _get_client() try: response = client.media.generate_upload_token() - upload_url = response.get("uploadUrl", "") - token = response.get("token", "") - expires_at = response.get("expiresAt", "") + upload_url = str(response.uploadUrl) if response.uploadUrl else "" + token = response.token or "" + expires_at = str(response.expiresAt) if response.expiresAt else "" return f"""📤 Upload link generated! @@ -702,25 +583,15 @@ def media_generate_upload_link() -> str: @mcp.tool() +@use_tool_def("media_check_upload_status") def media_check_upload_status(token: str) -> str: - """ - Check the status of an upload token and get uploaded file URLs. - - Use this after the user has uploaded files through the browser upload page. - - Args: - token: The upload token from media_generate_upload_link - - Returns: - Status and uploaded file URLs if completed - """ client = _get_client() try: response = client.media.check_upload_token(token) - status = response.get("status", "unknown") - files = response.get("files", []) + status = response.status.value if response.status else "unknown" + files = response.files or [] if status == "pending": return f"""⏳ Upload pending @@ -742,12 +613,12 @@ def media_check_upload_status(token: str) -> str: media_urls = [] for f in files: - url = f.get("url", "") + url = str(f.url) if f.url else "" media_urls.append(url) - lines.append(f"- {f.get('filename', 'unknown')}") - lines.append(f" Type: {f.get('type', 'N/A')}") + lines.append(f"- {f.filename or 'unknown'}") + lines.append(f" Type: {f.type.value if f.type else 'N/A'}") lines.append(f" URL: {url}") - lines.append(f" Size: {f.get('size', 0) / 1024:.1f} KB") + lines.append(f" Size: {(f.size or 0) / 1024:.1f} KB") lines.append("") lines.append( diff --git a/src/late/mcp/tool_definitions.py b/src/late/mcp/tool_definitions.py index 5202c17..a50f61e 100644 --- a/src/late/mcp/tool_definitions.py +++ b/src/late/mcp/tool_definitions.py @@ -1,16 +1,21 @@ """ Centralized tool definitions for MCP and documentation. -This file is the single source of truth for tool parameters and descriptions. +This file is the SINGLE SOURCE OF TRUTH for tool parameters and descriptions. Used by: -- MCP server (server.py) for tool definitions -- Documentation generation (can be exported to MDX) +- MCP server (server.py) for tool definitions via @use_tool_def decorator +- Documentation generation (generate_docs.py for MDX output) + +To update tool documentation, edit ONLY this file. """ from __future__ import annotations -from dataclasses import dataclass -from typing import Any +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Callable @dataclass @@ -35,175 +40,625 @@ class ToolDef: """Definition of a tool.""" name: str - description: str - params: list[ParamDef] + summary: str # Short one-line description + description: str # Detailed description with usage guidelines + params: list[ParamDef] = field(default_factory=list) + + def get_docstring(self) -> str: + """Generate docstring for MCP function.""" + lines = [self.description, ""] + + if self.params: + lines.append("Args:") + for param in self.params: + req = " (required)" if param.required else "" + default = ( + f" Default: {param.default}." + if param.default is not None and not param.required + else "" + ) + lines.append(f" {param.name}: {param.description}{req}{default}") + + return "\n".join(lines) def to_mdx_section(self) -> str: """Generate MDX documentation section.""" lines = [ - f"### {self.name}", + f"### `{self.name}`", + "", + self.summary, "", self.description, "", - "| Parameter | Type | Description | Required | Default |", - "|-----------|------|-------------|----------|---------|", ] - lines.extend(p.to_mdx_row() for p in self.params) + + if self.params: + lines.extend([ + "| Parameter | Type | Description | Required | Default |", + "|-----------|------|-------------|----------|---------|", + ]) + lines.extend(p.to_mdx_row() for p in self.params) + lines.append("") + return "\n".join(lines) # ============================================================================= -# POSTS TOOL DEFINITIONS +# DECORATOR FOR APPLYING TOOL DEFINITIONS +# ============================================================================= + + +def use_tool_def(tool_name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: + """ + Decorator to apply tool definition docstring to a function. + + Usage: + @mcp.tool() + @use_tool_def("posts_create") + def posts_create(...): + ... # Implementation only, no docstring needed + """ + + def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + tool = TOOL_DEFINITIONS.get(tool_name) + if tool: + # Set docstring on the original function + func.__doc__ = tool.get_docstring() + # Return the function directly - no wrapper needed + # The @mcp.tool() decorator will read __doc__ from this function + return func + + return decorator + + +# ============================================================================= +# ACCOUNTS TOOLS +# ============================================================================= + +ACCOUNTS_LIST = ToolDef( + name="accounts_list", + summary="List all connected social media accounts.", + description="""List all connected social media accounts. + +Returns the platform, username, and account ID for each connected account. +Use this to find account IDs needed for creating posts.""", + params=[], +) + +ACCOUNTS_GET = ToolDef( + name="accounts_get", + summary="Get account details for a specific platform.", + description="""Get account details for a specific platform. + +Returns username and ID for the first account matching the platform.""", + params=[ + ParamDef( + name="platform", + type="str", + description="Platform name: twitter, instagram, linkedin, tiktok, bluesky, facebook, youtube, pinterest, threads", + required=True, + ), + ], +) + +# ============================================================================= +# PROFILES TOOLS +# ============================================================================= + +PROFILES_LIST = ToolDef( + name="profiles_list", + summary="List all profiles.", + description="""List all profiles. + +Profiles group multiple social accounts together for easier management.""", + params=[], +) + +PROFILES_GET = ToolDef( + name="profiles_get", + summary="Get details of a specific profile.", + description="Get details of a specific profile including name, description, and color.", + params=[ + ParamDef( + name="profile_id", + type="str", + description="The profile ID", + required=True, + ), + ], +) + +PROFILES_CREATE = ToolDef( + name="profiles_create", + summary="Create a new profile.", + description="Create a new profile for grouping social accounts.", + params=[ + ParamDef( + name="name", + type="str", + description="Profile name", + required=True, + ), + ParamDef( + name="description", + type="str", + description="Optional description", + required=False, + default="", + ), + ParamDef( + name="color", + type="str", + description="Optional hex color (e.g., '#4CAF50')", + required=False, + default="", + ), + ], +) + +PROFILES_UPDATE = ToolDef( + name="profiles_update", + summary="Update an existing profile.", + description="Update an existing profile. Only provided fields will be changed.", + params=[ + ParamDef( + name="profile_id", + type="str", + description="The profile ID to update", + required=True, + ), + ParamDef( + name="name", + type="str", + description="New name (leave empty to keep current)", + required=False, + default="", + ), + ParamDef( + name="description", + type="str", + description="New description (leave empty to keep current)", + required=False, + default="", + ), + ParamDef( + name="color", + type="str", + description="New hex color (leave empty to keep current)", + required=False, + default="", + ), + ParamDef( + name="is_default", + type="bool", + description="Set as default profile", + required=False, + default=False, + ), + ], +) + +PROFILES_DELETE = ToolDef( + name="profiles_delete", + summary="Delete a profile.", + description="Delete a profile. The profile must have no connected accounts.", + params=[ + ParamDef( + name="profile_id", + type="str", + description="The profile ID to delete", + required=True, + ), + ], +) + +# ============================================================================= +# POSTS TOOLS # ============================================================================= -POSTS_CREATE_PARAMS = [ - ParamDef( - name="content", - type="str", - description="The post content/text", - required=True, - ), - ParamDef( - name="platform", - type="str", - description="Target platform: twitter, instagram, linkedin, tiktok, bluesky, facebook, youtube, pinterest, threads", - required=True, - ), - ParamDef( - name="is_draft", - type="bool", - description="Save as draft without scheduling. Draft posts can be edited and scheduled later", - required=False, - default=False, - ), - ParamDef( - name="publish_now", - type="bool", - description="Publish immediately instead of scheduling", - required=False, - default=False, - ), - ParamDef( - name="schedule_minutes", - type="int", - description="Minutes from now to schedule the post. Ignored if publish_now=True or is_draft=True", - required=False, - default=60, - ), - ParamDef( - name="media_urls", - type="str", - description="Comma-separated URLs of media files to attach (images, videos, GIFs)", - required=False, - default="", - ), - ParamDef( - name="title", - type="str", - description="Optional title (required for YouTube, recommended for Pinterest)", - required=False, - default="", - ), -] +POSTS_LIST = ToolDef( + name="posts_list", + summary="List posts with optional filtering.", + description="""List posts with optional filtering by status. + +Status options: draft, scheduled, published, failed""", + params=[ + ParamDef( + name="status", + type="str", + description="Filter by status: draft, scheduled, published, failed. Leave empty for all posts", + required=False, + default="", + ), + ParamDef( + name="limit", + type="int", + description="Maximum number of posts to return", + required=False, + default=10, + ), + ], +) + +POSTS_GET = ToolDef( + name="posts_get", + summary="Get details of a specific post.", + description="Get full details of a specific post including content, status, and scheduling info.", + params=[ + ParamDef( + name="post_id", + type="str", + description="The post ID to retrieve", + required=True, + ), + ], +) POSTS_CREATE = ToolDef( name="posts_create", - description="""Create a new social media post. - -**Scheduling behavior:** -- `is_draft=True`: Save as draft (no scheduling, can edit later) -- `publish_now=True`: Publish immediately -- Neither: Schedule for `schedule_minutes` from now (default: 60 min)""", - params=POSTS_CREATE_PARAMS, -) - -POSTS_CROSS_POST_PARAMS = [ - ParamDef( - name="content", - type="str", - description="The post content/text", - required=True, - ), - ParamDef( - name="platforms", - type="str", - description="Comma-separated list of platforms (e.g., 'twitter,linkedin,bluesky')", - required=True, - ), - ParamDef( - name="is_draft", - type="bool", - description="Save as draft without scheduling", - required=False, - default=False, - ), - ParamDef( - name="publish_now", - type="bool", - description="Publish immediately instead of scheduling", - required=False, - default=False, - ), - ParamDef( - name="media_urls", - type="str", - description="Comma-separated URLs of media files to attach", - required=False, - default="", - ), -] + summary="Create a social media post (draft, scheduled, or immediate).", + description="""Create a social media post. Can be saved as DRAFT, SCHEDULED, or PUBLISHED immediately. + +⚠️ IMPORTANT - Choose the correct mode based on user intent: + +**DRAFT MODE (is_draft=True)** +Use when user says: "draft", "borrador", "save for later", "don't publish", "save it", "guardar" +→ Post is saved but NOT published and NOT scheduled. User can edit it later. + +**IMMEDIATE MODE (publish_now=True)** +Use when user says: "publish now", "post now", "publica ya", "immediately", "right now", "ahora" +→ Post goes live IMMEDIATELY. + +**SCHEDULED MODE (default)** +Use when user says: "schedule", "programar", "in X minutes/hours", "at 3pm", "tomorrow" +→ Post is scheduled for future publication. Use schedule_minutes to set the delay. + +Examples: +- "Create a draft tweet" → is_draft=True +- "Post this to Twitter now" → publish_now=True +- "Schedule a LinkedIn post for 2 hours from now" → schedule_minutes=120""", + params=[ + ParamDef( + name="content", + type="str", + description="The post text/content", + required=True, + ), + ParamDef( + name="platform", + type="str", + description="Target platform: twitter, instagram, linkedin, tiktok, bluesky, facebook, youtube, pinterest, threads", + required=True, + ), + ParamDef( + name="is_draft", + type="bool", + description="Set to True to save as DRAFT (not published, not scheduled). Use when user wants to save without publishing", + required=False, + default=False, + ), + ParamDef( + name="publish_now", + type="bool", + description="Set to True to publish IMMEDIATELY. Post goes live right now", + required=False, + default=False, + ), + ParamDef( + name="schedule_minutes", + type="int", + description="Minutes from now to schedule. Only used when is_draft=False AND publish_now=False", + required=False, + default=60, + ), + ParamDef( + name="media_urls", + type="str", + description="Comma-separated URLs of media files to attach (images, videos)", + required=False, + default="", + ), + ParamDef( + name="title", + type="str", + description="Post title (required for YouTube, recommended for Pinterest)", + required=False, + default="", + ), + ], +) + +POSTS_PUBLISH_NOW = ToolDef( + name="posts_publish_now", + summary="Publish a post immediately.", + description="""Publish a post immediately to a platform. The post goes live right away. + +Use this when user explicitly wants to publish NOW, not schedule for later. +This is a convenience wrapper around posts_create with publish_now=True.""", + params=[ + ParamDef( + name="content", + type="str", + description="The post text/content", + required=True, + ), + ParamDef( + name="platform", + type="str", + description="Target platform: twitter, instagram, linkedin, tiktok, bluesky, etc.", + required=True, + ), + ParamDef( + name="media_urls", + type="str", + description="Comma-separated URLs of media files to attach", + required=False, + default="", + ), + ], +) POSTS_CROSS_POST = ToolDef( name="posts_cross_post", - description="Post the same content to multiple platforms at once.", - params=POSTS_CROSS_POST_PARAMS, -) - -POSTS_LIST_PARAMS = [ - ParamDef( - name="status", - type="str", - description="Filter by status: draft, scheduled, published, failed. Empty for all", - required=False, - default="", - ), - ParamDef( - name="limit", - type="int", - description="Maximum number of posts to return", - required=False, - default=10, - ), -] + summary="Post the same content to multiple platforms.", + description="""Post the same content to multiple platforms at once. -POSTS_LIST = ToolDef( - name="posts_list", - description="List posts with optional filtering by status.", - params=POSTS_LIST_PARAMS, +⚠️ IMPORTANT - Choose the correct mode based on user intent: + +**DRAFT MODE (is_draft=True)** +Use when user says: "draft", "borrador", "save for later", "don't publish" +→ Posts are saved but NOT published. User can edit them later. + +**IMMEDIATE MODE (publish_now=True)** +Use when user says: "publish now", "post now", "immediately" +→ Posts go live IMMEDIATELY on all platforms. + +**SCHEDULED MODE (default)** +Use when user says: "schedule", "programar", "in X hours" +→ Posts are scheduled for 1 hour from now.""", + params=[ + ParamDef( + name="content", + type="str", + description="The post text/content", + required=True, + ), + ParamDef( + name="platforms", + type="str", + description="Comma-separated list of platforms (e.g., 'twitter,linkedin,bluesky')", + required=True, + ), + ParamDef( + name="is_draft", + type="bool", + description="Set to True to save as DRAFT (not published). Use when user wants to save without publishing", + required=False, + default=False, + ), + ParamDef( + name="publish_now", + type="bool", + description="Set to True to publish IMMEDIATELY to all platforms", + required=False, + default=False, + ), + ParamDef( + name="media_urls", + type="str", + description="Comma-separated URLs of media files to attach", + required=False, + default="", + ), + ], +) + +POSTS_UPDATE = ToolDef( + name="posts_update", + summary="Update an existing post.", + description="""Update an existing post. + +Only draft, scheduled, and failed posts can be updated. +Published posts cannot be modified.""", + params=[ + ParamDef( + name="post_id", + type="str", + description="The post ID to update", + required=True, + ), + ParamDef( + name="content", + type="str", + description="New content (leave empty to keep current)", + required=False, + default="", + ), + ParamDef( + name="scheduled_for", + type="str", + description="New schedule time as ISO string (leave empty to keep current)", + required=False, + default="", + ), + ParamDef( + name="title", + type="str", + description="New title (leave empty to keep current)", + required=False, + default="", + ), + ], +) + +POSTS_DELETE = ToolDef( + name="posts_delete", + summary="Delete a post.", + description="""Delete a post by ID. + +Published posts cannot be deleted.""", + params=[ + ParamDef( + name="post_id", + type="str", + description="The post ID to delete", + required=True, + ), + ], +) + +POSTS_RETRY = ToolDef( + name="posts_retry", + summary="Retry a failed post.", + description="Retry publishing a failed post. Only works on posts with 'failed' status.", + params=[ + ParamDef( + name="post_id", + type="str", + description="The ID of the failed post to retry", + required=True, + ), + ], +) + +POSTS_LIST_FAILED = ToolDef( + name="posts_list_failed", + summary="List all failed posts.", + description="List all failed posts that can be retried.", + params=[ + ParamDef( + name="limit", + type="int", + description="Maximum number of posts to return", + required=False, + default=10, + ), + ], ) +POSTS_RETRY_ALL_FAILED = ToolDef( + name="posts_retry_all_failed", + summary="Retry all failed posts.", + description="Retry all failed posts at once.", + params=[], +) + +# ============================================================================= +# MEDIA TOOLS +# ============================================================================= + +MEDIA_GENERATE_UPLOAD_LINK = ToolDef( + name="media_generate_upload_link", + summary="Generate an upload URL for media files.", + description="""Generate a unique upload URL for the user to upload files via browser. + +Use this when the user wants to include images or videos in their post. +The flow is: +1. Call this tool to get an upload URL +2. Ask the user to open the URL in their browser +3. User uploads files through the web interface +4. Call media_check_upload_status to get the uploaded file URLs +5. Use those URLs when creating the post with posts_create""", + params=[], +) + +MEDIA_CHECK_UPLOAD_STATUS = ToolDef( + name="media_check_upload_status", + summary="Check upload status and get file URLs.", + description="""Check the status of an upload token and get uploaded file URLs. + +Use this after the user has uploaded files through the browser upload page. +Returns: pending (waiting for upload), completed (files ready), or expired (token expired).""", + params=[ + ParamDef( + name="token", + type="str", + description="The upload token from media_generate_upload_link", + required=True, + ), + ], +) + + # ============================================================================= -# ALL TOOL DEFINITIONS +# ALL TOOL DEFINITIONS REGISTRY # ============================================================================= -TOOL_DEFINITIONS = { +TOOL_DEFINITIONS: dict[str, ToolDef] = { + # Accounts + "accounts_list": ACCOUNTS_LIST, + "accounts_get": ACCOUNTS_GET, + # Profiles + "profiles_list": PROFILES_LIST, + "profiles_get": PROFILES_GET, + "profiles_create": PROFILES_CREATE, + "profiles_update": PROFILES_UPDATE, + "profiles_delete": PROFILES_DELETE, + # Posts + "posts_list": POSTS_LIST, + "posts_get": POSTS_GET, "posts_create": POSTS_CREATE, + "posts_publish_now": POSTS_PUBLISH_NOW, "posts_cross_post": POSTS_CROSS_POST, - "posts_list": POSTS_LIST, + "posts_update": POSTS_UPDATE, + "posts_delete": POSTS_DELETE, + "posts_retry": POSTS_RETRY, + "posts_list_failed": POSTS_LIST_FAILED, + "posts_retry_all_failed": POSTS_RETRY_ALL_FAILED, + # Media + "media_generate_upload_link": MEDIA_GENERATE_UPLOAD_LINK, + "media_check_upload_status": MEDIA_CHECK_UPLOAD_STATUS, } -def generate_mdx_docs() -> str: - """Generate complete MDX documentation for all tools.""" +# ============================================================================= +# DOCUMENTATION GENERATION +# ============================================================================= + + +def generate_mdx_tools_reference() -> str: + """Generate MDX documentation for all tools.""" sections = [ "## Tool Reference", "", "Detailed parameters for each MCP tool.", "", ] - for tool in TOOL_DEFINITIONS.values(): - sections.append(tool.to_mdx_section()) + + # Group by category + categories = { + "Accounts": ["accounts_list", "accounts_get"], + "Profiles": [ + "profiles_list", + "profiles_get", + "profiles_create", + "profiles_update", + "profiles_delete", + ], + "Posts": [ + "posts_create", + "posts_publish_now", + "posts_cross_post", + "posts_list", + "posts_get", + "posts_update", + "posts_delete", + "posts_retry", + "posts_list_failed", + "posts_retry_all_failed", + ], + "Media": ["media_generate_upload_link", "media_check_upload_status"], + } + + for category, tool_names in categories.items(): + sections.append(f"### {category}") + sections.append("") + for name in tool_names: + tool = TOOL_DEFINITIONS.get(name) + if tool: + sections.append(tool.to_mdx_section()) sections.append("") + return "\n".join(sections) @@ -212,15 +667,4 @@ def get_tool_docstring(tool_name: str) -> str: tool = TOOL_DEFINITIONS.get(tool_name) if not tool: return "" - - lines = [tool.description, "", "Args:"] - for param in tool.params: - req = " (required)" if param.required else "" - default = ( - f" (default: {param.default})" - if param.default is not None and not param.required - else "" - ) - lines.append(f" {param.name}: {param.description}{req}{default}") - - return "\n".join(lines) + return tool.get_docstring() diff --git a/src/late/models/_generated/models.py b/src/late/models/_generated/models.py index b260cdf..07cc92f 100644 --- a/src/late/models/_generated/models.py +++ b/src/late/models/_generated/models.py @@ -1,18 +1,18 @@ # generated by datamodel-codegen: # filename: public-api.yaml -# timestamp: 2025-12-15T13:54:40+00:00 +# timestamp: 2025-12-15T16:41:41+00:00 from __future__ import annotations from enum import Enum -from typing import Annotated, Any +from typing import Annotated, Any, Dict, List from pydantic import AnyUrl, AwareDatetime, BaseModel, Field class ErrorResponse(BaseModel): error: str | None = None - details: dict[str, Any] | None = None + details: Dict[str, Any] | None = None class Type(Enum): @@ -80,11 +80,11 @@ class Visibility(Enum): class ThreadItem(BaseModel): content: str | None = None - mediaItems: list[MediaItem] | None = None + mediaItems: List[MediaItem] | None = None class TwitterPlatformData(BaseModel): - threadItems: list[ThreadItem] | None = None + threadItems: List[ThreadItem] | None = None """ Sequence of tweets in a thread. First item is the root tweet. """ @@ -101,7 +101,7 @@ class ThreadsPlatformData(BaseModel): """ - threadItems: list[ThreadItem] | None = None + threadItems: List[ThreadItem] | None = None """ Sequence of posts in a Threads thread (root then replies in order). """ @@ -190,7 +190,7 @@ class InstagramPlatformData(BaseModel): """ For Reels only. When true (default), the Reel appears on both the Reels tab and your main profile feed. Set to false to post to the Reels tab only. """ - collaborators: list[str] | None = None + collaborators: List[str] | None = None """ Up to 3 Instagram usernames to invite as collaborators (feed/Reels only) """ @@ -198,7 +198,7 @@ class InstagramPlatformData(BaseModel): """ Optional first comment to add after the post is created (not applied to Stories) """ - userTags: list[UserTag] | None = None + userTags: List[UserTag] | None = None """ Tag Instagram users in photos by username and position coordinates. Only works for single image posts and the first image of carousel posts. Not supported for stories or videos. """ @@ -453,7 +453,7 @@ class QueueSchedule(BaseModel): """ IANA timezone (e.g., America/New_York) """ - slots: list[QueueSlot] | None = None + slots: List[QueueSlot] | None = None active: bool | None = None """ Whether the queue is active @@ -482,7 +482,7 @@ class Profile(BaseModel): class SocialAccount(BaseModel): field_id: Annotated[str | None, Field(alias="_id")] = None platform: str | None = None - profileId: str | None = None + profileId: str | Profile | None = None username: str | None = None displayName: str | None = None isActive: bool | None = None @@ -553,114 +553,6 @@ class UsageStats(BaseModel): usage: Usage | None = None -class Status1(Enum): - PROCESSING = "processing" - COMPLETED = "completed" - FAILED = "failed" - - -class VideoClip(BaseModel): - index: Annotated[int | None, Field(examples=[1])] = None - url: Annotated[ - AnyUrl | None, Field(examples=["https://clips.example.com/clip-1-clean.mp4"]) - ] = None - """ - Clean version URL (default) - """ - duration: Annotated[float | None, Field(examples=[58.3])] = None - """ - Clip duration in seconds - """ - start_time: Annotated[float | None, Field(examples=[120.5])] = None - """ - Start time in original video (seconds) - """ - end_time: Annotated[float | None, Field(examples=[178.8])] = None - """ - End time in original video (seconds) - """ - watermarkUrl: Annotated[ - AnyUrl | None, - Field(examples=["https://clips.example.com/clip-1-watermark.mp4"]), - ] = None - """ - Watermarked version URL (free tier) - """ - cleanUrl: Annotated[ - AnyUrl | None, Field(examples=["https://clips.example.com/clip-1-clean.mp4"]) - ] = None - """ - Clean version URL (addon users) - """ - - -class Status2(Enum): - PROCESSING = "processing" - - -class VideoClipJobProcessing(BaseModel): - job_id: Annotated[str | None, Field(examples=["abc123def456"])] = None - status: Annotated[Status2 | None, Field(examples=["processing"])] = None - message: Annotated[ - str | None, - Field(examples=["Video is being processed. Check back in a few minutes."]), - ] = None - - -class Status3(Enum): - COMPLETED = "completed" - - -class VideoClipJobCompleted(BaseModel): - job_id: Annotated[str | None, Field(examples=["abc123def456"])] = None - status: Annotated[Status3 | None, Field(examples=["completed"])] = None - total_clips: Annotated[int | None, Field(examples=[5])] = None - clips: list[VideoClip] | None = None - - -class Status4(Enum): - FAILED = "failed" - - -class VideoClipJobFailed(BaseModel): - job_id: Annotated[str | None, Field(examples=["abc123def456"])] = None - status: Annotated[Status4 | None, Field(examples=["failed"])] = None - error: Annotated[ - str | None, - Field( - examples=[ - "Video has no audio track. Please use a video that contains both audio and video." - ] - ), - ] = None - - -class VideoClipUsageStats(BaseModel): - current: Annotated[int | None, Field(examples=[3])] = None - """ - Current usage this month - """ - limit: Annotated[int | None, Field(examples=[5])] = None - """ - Monthly limit - """ - remaining: Annotated[int | None, Field(examples=[2])] = None - """ - Remaining credits this month - """ - canCreate: Annotated[bool | None, Field(examples=[True])] = None - """ - Whether user can create new jobs - """ - hasAddon: Annotated[bool | None, Field(examples=[False])] = None - """ - Whether user has AI Clipping addon - """ - message: Annotated[str | None, Field(examples=["Free tier (with watermark)"])] = ( - None - ) - - class PostAnalytics(BaseModel): impressions: Annotated[int | None, Field(examples=[0])] = None reach: Annotated[int | None, Field(examples=[0])] = None @@ -708,7 +600,7 @@ class AnalyticsSinglePostResponse(BaseModel): scheduledFor: AwareDatetime | None = None publishedAt: AwareDatetime | None = None analytics: PostAnalytics | None = None - platformAnalytics: list[PlatformAnalytics] | None = None + platformAnalytics: List[PlatformAnalytics] | None = None platform: str | None = None platformPostUrl: AnyUrl | None = None isExternal: bool | None = None @@ -728,20 +620,20 @@ class Post1(BaseModel): publishedAt: AwareDatetime | None = None status: str | None = None analytics: PostAnalytics | None = None - platforms: list[PlatformAnalytics] | None = None + platforms: List[PlatformAnalytics] | None = None platform: str | None = None platformPostUrl: AnyUrl | None = None isExternal: bool | None = None thumbnailUrl: AnyUrl | None = None mediaType: MediaType1 | None = None - mediaItems: list[MediaItem] | None = None + mediaItems: List[MediaItem] | None = None class AnalyticsListResponse(BaseModel): overview: AnalyticsOverview | None = None - posts: list[Post1] | None = None + posts: List[Post1] | None = None pagination: Pagination | None = None - accounts: list[SocialAccount] | None = None + accounts: List[SocialAccount] | None = None """ Connected social accounts (followerCount and followersLastUpdated only included if user has analytics add-on) """ @@ -756,7 +648,7 @@ class PostDeleteResponse(BaseModel): class ProfilesListResponse(BaseModel): - profiles: list[Profile] | None = None + profiles: List[Profile] | None = None class ProfileGetResponse(BaseModel): @@ -778,7 +670,7 @@ class ProfileDeleteResponse(BaseModel): class AccountsListResponse(BaseModel): - accounts: list[SocialAccount] | None = None + accounts: List[SocialAccount] | None = None hasAnalyticsAccess: bool | None = None """ Whether user has analytics add-on access @@ -801,7 +693,7 @@ class Aggregation(Enum): class FollowerStatsResponse(BaseModel): - accounts: list[AccountWithFollowerStats] | None = None + accounts: List[AccountWithFollowerStats] | None = None dateRange: DateRange | None = None aggregation: Aggregation | None = None @@ -821,10 +713,10 @@ class UploadedFile(BaseModel): class MediaUploadResponse(BaseModel): - files: list[UploadedFile] | None = None + files: List[UploadedFile] | None = None -class Status5(Enum): +class Status1(Enum): PENDING = "pending" COMPLETED = "completed" EXPIRED = "expired" @@ -834,13 +726,13 @@ class UploadTokenResponse(BaseModel): token: str | None = None uploadUrl: AnyUrl | None = None expiresAt: AwareDatetime | None = None - status: Status5 | None = None + status: Status1 | None = None class UploadTokenStatusResponse(BaseModel): token: str | None = None - status: Status5 | None = None - files: list[UploadedFile] | None = None + status: Status1 | None = None + files: List[UploadedFile] | None = None createdAt: AwareDatetime | None = None expiresAt: AwareDatetime | None = None completedAt: AwareDatetime | None = None @@ -849,13 +741,13 @@ class UploadTokenStatusResponse(BaseModel): class QueueSlotsResponse(BaseModel): exists: bool | None = None schedule: QueueSchedule | None = None - nextSlots: list[AwareDatetime] | None = None + nextSlots: List[AwareDatetime] | None = None class QueueUpdateResponse(BaseModel): success: bool | None = None schedule: QueueSchedule | None = None - nextSlots: list[AwareDatetime] | None = None + nextSlots: List[AwareDatetime] | None = None reshuffledCount: int | None = None @@ -867,7 +759,7 @@ class QueueDeleteResponse(BaseModel): class QueuePreviewResponse(BaseModel): profileId: str | None = None count: int | None = None - slots: list[AwareDatetime] | None = None + slots: List[AwareDatetime] | None = None class QueueNextSlotResponse(BaseModel): @@ -889,7 +781,7 @@ class DownloadResponse(BaseModel): title: str | None = None thumbnail: AnyUrl | None = None duration: int | None = None - formats: list[DownloadFormat] | None = None + formats: List[DownloadFormat] | None = None class TranscriptSegment(BaseModel): @@ -900,11 +792,11 @@ class TranscriptSegment(BaseModel): class TranscriptResponse(BaseModel): transcript: str | None = None - segments: list[TranscriptSegment] | None = None + segments: List[TranscriptSegment] | None = None language: str | None = None -class Status7(Enum): +class Status3(Enum): SAFE = "safe" BANNED = "banned" RESTRICTED = "restricted" @@ -913,12 +805,12 @@ class Status7(Enum): class HashtagInfo(BaseModel): hashtag: str | None = None - status: Status7 | None = None + status: Status3 | None = None postCount: int | None = None class HashtagCheckResponse(BaseModel): - hashtags: list[HashtagInfo] | None = None + hashtags: List[HashtagInfo] | None = None class CaptionResponse(BaseModel): @@ -934,7 +826,7 @@ class User(BaseModel): class UsersListResponse(BaseModel): - users: list[User] | None = None + users: List[User] | None = None class UserGetResponse(BaseModel): @@ -950,35 +842,14 @@ class TikTokPlatformData(BaseModel): tiktokSettings: TikTokSettings | None = None -class VideoClipJob(BaseModel): - field_id: Annotated[ - str | None, Field(alias="_id", examples=["507f1f77bcf86cd799439011"]) - ] = None - jobId: Annotated[str | None, Field(examples=["abc123def456"])] = None - videoUrl: Annotated[ - AnyUrl | None, Field(examples=["https://storage.example.com/video.mp4"]) - ] = None - videoFileName: Annotated[str | None, Field(examples=["my-video.mp4"])] = None - status: Annotated[Status1 | None, Field(examples=["completed"])] = None - clips: list[VideoClip] | None = None - totalClips: Annotated[int | None, Field(examples=[5])] = None - error: Annotated[str | None, Field(examples=[None])] = None - createdAt: Annotated[ - AwareDatetime | None, Field(examples=["2025-10-22T10:30:00Z"]) - ] = None - completedAt: Annotated[ - AwareDatetime | None, Field(examples=["2025-10-22T10:45:00Z"]) - ] = None - - class PlatformTarget(BaseModel): platform: Annotated[str | None, Field(examples=["twitter"])] = None """ Supported values: twitter, threads, instagram, youtube, facebook, linkedin, pinterest, reddit, tiktok, bluesky, googlebusiness """ - accountId: str | None = None + accountId: str | SocialAccount | None = None customContent: str | None = None - customMedia: list[MediaItem] | None = None + customMedia: List[MediaItem] | None = None scheduledFor: AwareDatetime | None = None """ Optional per-platform scheduled time override (uses post.scheduledFor when omitted) @@ -1014,8 +885,8 @@ class PlatformTarget(BaseModel): ] = None """ Public URL of the published post on the platform. - Populated after successful publish. For immediate posts (publishNow=true), - this is included in the response. For scheduled posts, fetch the post + Populated after successful publish. For immediate posts (publishNow=true), + this is included in the response. For scheduled posts, fetch the post via GET /v1/posts/{postId} after the scheduled time. """ @@ -1027,19 +898,19 @@ class PlatformTarget(BaseModel): class Post(BaseModel): field_id: Annotated[str | None, Field(alias="_id")] = None - userId: str | None = None + userId: str | User | None = None title: str | None = None """ YouTube: title must be ≤ 100 characters. """ content: str | None = None - mediaItems: list[MediaItem] | None = None - platforms: list[PlatformTarget] | None = None + mediaItems: List[MediaItem] | None = None + platforms: List[PlatformTarget] | None = None scheduledFor: AwareDatetime | None = None timezone: str | None = None status: Status | None = None - tags: list[str] | None = None + tags: List[str] | None = None """ YouTube tag constraints when targeting YouTube: - No count cap; duplicates removed. @@ -1047,10 +918,10 @@ class Post(BaseModel): - Combined characters across all tags ≤ 500. """ - hashtags: list[str] | None = None - mentions: list[str] | None = None + hashtags: List[str] | None = None + mentions: List[str] | None = None visibility: Visibility | None = None - metadata: dict[str, Any] | None = None + metadata: Dict[str, Any] | None = None queuedFromProfile: str | None = None """ Profile ID if the post was scheduled via the queue @@ -1060,7 +931,7 @@ class Post(BaseModel): class PostsListResponse(BaseModel): - posts: list[Post] | None = None + posts: List[Post] | None = None pagination: Pagination | None = None diff --git a/uv.lock b/uv.lock index ef65568..754a336 100644 --- a/uv.lock +++ b/uv.lock @@ -680,7 +680,7 @@ wheels = [ [[package]] name = "late-sdk" -version = "1.1.0" +version = "1.1.1" source = { editable = "." } dependencies = [ { name = "httpx" },