github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-01-10 23:58:06 -05:00
Code Issues Packages Projects Releases Wiki Activity

feat(block): Add Ayrshare integration for social media posting (#9946)

Browse Source 
This PR implements a comprehensive Ayrshare social media integration for
AutoGPT Platform, enabling users to post content across multiple social
media platforms through a unified interface. Ayrshare provides a single
API to manage posts across Facebook, Twitter/X, LinkedIn, Instagram,
YouTube, TikTok, Pinterest, Reddit, Telegram, Google My Business,
Bluesky, Snapchat, and Threads.

The integration addresses the need for social media automation and
content distribution workflows within AutoGPT agents, allowing users to:
- Connect their social media accounts via SSO
- Post content with platform-specific options and constraints
- Schedule posts across multiple platforms simultaneously
- Handle platform-specific media requirements and validation

⚠️ To simplify the review process all except the twitter post block has
been commented out, future pr's will uncomment other platfroms so we can
test them in isolation.

### Changes 🏗️

#### Backend Integration (`backend/integrations/ayrshare.py`)
- **AyrshareClient**: Complete API client implementation with post
creation, profile management, and JWT generation
- **SocialPlatform enum**: Comprehensive platform definitions for all
supported social networks
- **Response models**: PostResponse, ProfileResponse, JWTResponse for
type-safe API interactions
- **Error handling**: Custom AyrshareAPIException with proper HTTP
status code handling

#### Social Media Posting Blocks (`backend/blocks/ayrshare/post.py`)
- **BaseAyrshareInput**: Shared input schema with common fields (post
text, media URLs, scheduling, etc.)
- **Platform-specific blocks**: 13 dedicated posting blocks, each with
platform-specific validation and options:
  - PostToFacebookBlock: Carousel, Reels, Stories, targeting, alt text
- PostToXBlock: Threads, polls, long posts, premium features, subtitles
- PostToLinkedInBlock: Document support, visibility controls, audience
targeting
  - PostToInstagramBlock: Stories, Reels, user tags, collaborators
- PostToYouTubeBlock: Video uploads, playlists, visibility, country
targeting
  - PostToPinterestBlock: Pins, carousels, board management
  - PostToTikTokBlock: Video/image posts, AI labeling, brand content
  - PostToRedditBlock: Basic posting functionality
  - PostToTelegramBlock: GIF handling, mentions
  - PostToGMBBlock: Event/offer posts, call-to-action buttons
  - PostToBlueskyBlock: Character limit validation, alt text
  - PostToSnapchatBlock: Story types, video thumbnails
  - PostToThreadsBlock: Hashtag restrictions, carousel support

#### Helper Models
- **CarouselItem**: Facebook carousel configuration
- **CallToAction, EventDetails, OfferDetails**: Google My Business post
types
- **InstagramUserTag**: Instagram user tagging with coordinates
- **LinkedInTargeting**: LinkedIn audience targeting options
- **PinterestCarouselOption**: Pinterest carousel image options
- **YouTubeTargeting**: YouTube country blocking/allowing

#### Authentication & SSO (`backend/server/integrations/router.py`)
- **SSO endpoint**: `/integrations/ayrshare/sso_url` for account linking
- **Profile management**: Automatic profile creation and key management
- **JWT generation**: Secure token generation for social media account
linking
- **Platform allowlist**: Configured access to all supported social
platforms

#### Frontend Integration (`frontend/src/components/CustomNode.tsx`)
- **AYRSHARE block type**: New BlockUIType.AYRSHARE for
Ayrshare-specific nodes
- **SSO button**: "Connect Social Media Accounts" with loading states
- **Handle generation**: Special handling for Ayrshare blocks with SSO
integration

#### Configuration
- **Environment variables**: Added AYRSHARE_API_KEY and AYRSHARE_JWT_KEY
to .env.example
- **Block registration**: All Ayrshare blocks registered in
AYRSHARE_NODE_IDS array

#### Type Safety & Error Handling
- **Modern typing**: Updated to use `list`, `dict`, `Any` instead of
legacy typing
- **Comprehensive validation**: Platform-specific constraints (character
limits, media counts, file types)
- **User-friendly errors**: Clear error messages for validation failures
and API errors

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:

  **Test Plan:**
  
  **Backend API Testing:**
  - [x] Verify AyrshareClient initializes correctly with API key
  - [x] Test JWT generation for SSO authentication
  - [x] Test profile creation and management
  - [x] Verify all 13 posting blocks are properly registered
  - [x] Test platform-specific validation rules for each block
  - [x] Verify error handling for missing credentials and API failures
  
  **Frontend Integration Testing:**
  - [x] Verify AYRSHARE block type renders correctly in flow editor
  - [x] Test SSO button functionality and popup window behavior
  - [x] Confirm loading states work properly during authentication
  - [x] Verify input handles generate correctly for Ayrshare blocks
  - [x] Test platform-specific input fields and validation
  
  **End-to-End Workflow Testing:**
  - [x] Create agent with Ayrshare posting blocks
  - [x] Test SSO flow: click "Connect Social Media Accounts" button
  - [x] Verify popup opens with Ayrshare authentication page
  - [x] Test social media account linking process
  - [x] Create posts with various platform-specific options:
      - [ X ] X (Twitter) - tested basic posting with image
  - [] Test scheduling functionality across platforms
  - [x] Verify media upload constraints and validation
  - [] Test error handling for invalid inputs and failed posts
  
  **Error Case Testing:**
  - [] Test behavior with missing AYRSHARE_API_KEY configuration
  - [] Test invalid social media credentials handling
  - [] Test network failure scenarios
  - [] Verify platform-specific validation error messages
  - [] Test character limit enforcement per platform
  - [] Test media file type and size restrictions
  
  **Security Testing:**
  - [ x ] Verify JWT tokens are properly generated and validated
  - [x] Test profile key isolation between users
  - [x] Confirm sensitive credentials are not logged
  - [x] Verify SSO popup prevents XSS attacks

#### For configuration changes:
- [x] `.env.example` is updated or already compatible with my changes
- [x] `docker-compose.yml` is updated or already compatible with my
changes
- [x] I have included a list of my configuration changes in the PR
description (under **Changes**)

**Configuration Changes:**
- Added `AYRSHARE_API_KEY` environment variable for Ayrshare API
authentication
- Added `AYRSHARE_JWT_KEY` environment variable for SSO token generation
- No docker-compose.yml changes required (uses existing backend
services)

---------

Co-authored-by: Reinier van der Leer <pwuts@agpt.co>
This commit is contained in:
Swifty
2025-07-25 15:33:29 +02:00
committed by GitHub
parent c955e9a4d7
commit 39fe22f7e7
34 changed files with 3884 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
autogpt_platform/backend/.env.example
View File

@@ -197,6 +197,10 @@ SMARTLEAD_API_KEY=
# ZeroBounce
ZEROBOUNCE_API_KEY=
# Ayrshare
AYRSHARE_API_KEY=
AYRSHARE_JWT_KEY=
## ===== OPTIONAL API KEYS END ===== ##
# Block Error Rate Monitoring

15
autogpt_platform/backend/backend/blocks/ayrshare/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
AYRSHARE_BLOCK_IDS = [
"cbd52c2a-06d2-43ed-9560-6576cc163283", # PostToBlueskyBlock
"3352f512-3524-49ed-a08f-003042da2fc1", # PostToFacebookBlock
"9e8f844e-b4a5-4b25-80f2-9e1dd7d67625", # PostToXBlock
"589af4e4-507f-42fd-b9ac-a67ecef25811", # PostToLinkedInBlock
"89b02b96-a7cb-46f4-9900-c48b32fe1552", # PostToInstagramBlock
"0082d712-ff1b-4c3d-8a8d-6c7721883b83", # PostToYouTubeBlock
"c7733580-3c72-483e-8e47-a8d58754d853", # PostToRedditBlock
"47bc74eb-4af2-452c-b933-af377c7287df", # PostToTelegramBlock
"2c38c783-c484-4503-9280-ef5d1d345a7e", # PostToGMBBlock
"3ca46e05-dbaa-4afb-9e95-5a429c4177e6", # PostToPinterestBlock
"7faf4b27-96b0-4f05-bf64-e0de54ae74e1", # PostToTikTokBlock
"f8c3b2e1-9d4a-4e5f-8c7b-6a9e8d2f1c3b", # PostToThreadsBlock
"a9d7f854-2c83-4e96-b3a1-7f2e9c5d4b8e", # PostToSnapchatBlock
]

144
autogpt_platform/backend/backend/blocks/ayrshare/_util.py Normal file
View File

@@ -0,0 +1,144 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from backend.data.block import BlockSchema
from backend.data.model import SchemaField
from backend.integrations.ayrshare import AyrshareClient
from backend.util.exceptions import MissingConfigError
class BaseAyrshareInput(BlockSchema):
"""Base input model for Ayrshare social media posts with common fields."""
post: str = SchemaField(
description="The post text to be published", default="", advanced=False
)
media_urls: list[str] = SchemaField(
description="Optional list of media URLs to include. Set is_video in advanced settings to true if you want to upload videos.",
default_factory=list,
advanced=False,
)
is_video: bool = SchemaField(
description="Whether the media is a video", default=False, advanced=True
)
schedule_date: Optional[datetime] = SchemaField(
description="UTC datetime for scheduling (YYYY-MM-DDThh:mm:ssZ)",
default=None,
advanced=True,
)
disable_comments: bool = SchemaField(
description="Whether to disable comments", default=False, advanced=True
)
shorten_links: bool = SchemaField(
description="Whether to shorten links", default=False, advanced=True
)
unsplash: Optional[str] = SchemaField(
description="Unsplash image configuration", default=None, advanced=True
)
requires_approval: bool = SchemaField(
description="Whether to enable approval workflow",
default=False,
advanced=True,
)
random_post: bool = SchemaField(
description="Whether to generate random post text",
default=False,
advanced=True,
)
random_media_url: bool = SchemaField(
description="Whether to generate random media", default=False, advanced=True
)
notes: Optional[str] = SchemaField(
description="Additional notes for the post", default=None, advanced=True
)
class CarouselItem(BaseModel):
"""Model for Facebook carousel items."""
name: str = Field(..., description="The name of the item")
link: str = Field(..., description="The link of the item")
picture: str = Field(..., description="The picture URL of the item")
class CallToAction(BaseModel):
"""Model for Google My Business Call to Action."""
action_type: str = Field(
..., description="Type of action (book, order, shop, learn_more, sign_up, call)"
)
url: Optional[str] = Field(
description="URL for the action (not required for 'call' action)"
)
class EventDetails(BaseModel):
"""Model for Google My Business Event details."""
title: str = Field(..., description="Event title")
start_date: str = Field(..., description="Event start date (ISO format)")
end_date: str = Field(..., description="Event end date (ISO format)")
class OfferDetails(BaseModel):
"""Model for Google My Business Offer details."""
title: str = Field(..., description="Offer title")
start_date: str = Field(..., description="Offer start date (ISO format)")
end_date: str = Field(..., description="Offer end date (ISO format)")
coupon_code: str = Field(..., description="Coupon code (max 58 characters)")
redeem_online_url: str = Field(..., description="URL to redeem the offer")
terms_conditions: str = Field(..., description="Terms and conditions")
class InstagramUserTag(BaseModel):
"""Model for Instagram user tags."""
username: str = Field(..., description="Instagram username (without @)")
x: Optional[float] = Field(description="X coordinate (0.0-1.0) for image posts")
y: Optional[float] = Field(description="Y coordinate (0.0-1.0) for image posts")
class LinkedInTargeting(BaseModel):
"""Model for LinkedIn audience targeting."""
countries: Optional[list[str]] = Field(
description="Country codes (e.g., ['US', 'IN', 'DE', 'GB'])"
)
seniorities: Optional[list[str]] = Field(
description="Seniority levels (e.g., ['Senior', 'VP'])"
)
degrees: Optional[list[str]] = Field(description="Education degrees")
fields_of_study: Optional[list[str]] = Field(description="Fields of study")
industries: Optional[list[str]] = Field(description="Industry categories")
job_functions: Optional[list[str]] = Field(description="Job function categories")
staff_count_ranges: Optional[list[str]] = Field(description="Company size ranges")
class PinterestCarouselOption(BaseModel):
"""Model for Pinterest carousel image options."""
title: Optional[str] = Field(description="Image title")
link: Optional[str] = Field(description="External destination link for the image")
description: Optional[str] = Field(description="Image description")
class YouTubeTargeting(BaseModel):
"""Model for YouTube country targeting."""
block: Optional[list[str]] = Field(
description="Country codes to block (e.g., ['US', 'CA'])"
)
allow: Optional[list[str]] = Field(
description="Country codes to allow (e.g., ['GB', 'AU'])"
)
def create_ayrshare_client():
"""Create an Ayrshare client instance."""
try:
return AyrshareClient()
except MissingConfigError:
return None

113
autogpt_platform/backend/backend/blocks/ayrshare/post_to_bluesky.py Normal file
View File

@@ -0,0 +1,113 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToBlueskyBlock(Block):
"""Block for posting to Bluesky with Bluesky-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Bluesky posts."""
# Override post field to include character limit information
post: str = SchemaField(
description="The post text to be published (max 300 characters for Bluesky)",
default="",
advanced=False,
)
# Override media_urls to include Bluesky-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs to include. Bluesky supports up to 4 images or 1 video.",
default_factory=list,
advanced=False,
)
# Bluesky-specific options
alt_text: list[str] = SchemaField(
description="Alt text for each media item (accessibility)",
default_factory=list,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="cbd52c2a-06d2-43ed-9560-6576cc163283",
description="Post to Bluesky using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToBlueskyBlock.Input,
output_schema=PostToBlueskyBlock.Output,
)
async def run(
self,
input_data: "PostToBlueskyBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Bluesky with Bluesky-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate character limit for Bluesky
if len(input_data.post) > 300:
yield "error", f"Post text exceeds Bluesky's 300 character limit ({len(input_data.post)} characters)"
return
# Validate media constraints for Bluesky
if len(input_data.media_urls) > 4:
yield "error", "Bluesky supports a maximum of 4 images or 1 video"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Bluesky-specific options
bluesky_options = {}
if input_data.alt_text:
bluesky_options["altText"] = input_data.alt_text
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.BLUESKY],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
bluesky_options=bluesky_options if bluesky_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

207
autogpt_platform/backend/backend/blocks/ayrshare/post_to_facebook.py Normal file
View File

@@ -0,0 +1,207 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, CarouselItem, create_ayrshare_client
class PostToFacebookBlock(Block):
"""Block for posting to Facebook with Facebook-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Facebook posts."""
# Facebook-specific options
is_carousel: bool = SchemaField(
description="Whether to post a carousel", default=False, advanced=True
)
carousel_link: str = SchemaField(
description="The URL for the 'See More At' button in the carousel",
default="",
advanced=True,
)
carousel_items: list[CarouselItem] = SchemaField(
description="List of carousel items with name, link and picture URLs. Min 2, max 10 items.",
default_factory=list,
advanced=True,
)
is_reels: bool = SchemaField(
description="Whether to post to Facebook Reels",
default=False,
advanced=True,
)
reels_title: str = SchemaField(
description="Title for the Reels video (max 255 chars)",
default="",
advanced=True,
)
reels_thumbnail: str = SchemaField(
description="Thumbnail URL for Reels video (JPEG/PNG, <10MB)",
default="",
advanced=True,
)
is_story: bool = SchemaField(
description="Whether to post as a Facebook Story",
default=False,
advanced=True,
)
media_captions: list[str] = SchemaField(
description="Captions for each media item",
default_factory=list,
advanced=True,
)
location_id: str = SchemaField(
description="Facebook Page ID or name for location tagging",
default="",
advanced=True,
)
age_min: int = SchemaField(
description="Minimum age for audience targeting (13,15,18,21,25)",
default=0,
advanced=True,
)
target_countries: list[str] = SchemaField(
description="List of country codes to target (max 25)",
default_factory=list,
advanced=True,
)
alt_text: list[str] = SchemaField(
description="Alt text for each media item",
default_factory=list,
advanced=True,
)
video_title: str = SchemaField(
description="Title for video post", default="", advanced=True
)
video_thumbnail: str = SchemaField(
description="Thumbnail URL for video post", default="", advanced=True
)
is_draft: bool = SchemaField(
description="Save as draft in Meta Business Suite",
default=False,
advanced=True,
)
scheduled_publish_date: str = SchemaField(
description="Schedule publish time in Meta Business Suite (UTC)",
default="",
advanced=True,
)
preview_link: str = SchemaField(
description="URL for custom link preview", default="", advanced=True
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="3352f512-3524-49ed-a08f-003042da2fc1",
description="Post to Facebook using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToFacebookBlock.Input,
output_schema=PostToFacebookBlock.Output,
)
async def run(
self,
input_data: "PostToFacebookBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Facebook with Facebook-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Facebook-specific options
facebook_options = {}
if input_data.is_carousel:
facebook_options["isCarousel"] = True
if input_data.carousel_link:
facebook_options["carouselLink"] = input_data.carousel_link
if input_data.carousel_items:
facebook_options["carouselItems"] = [
item.dict() for item in input_data.carousel_items
]
if input_data.is_reels:
facebook_options["isReels"] = True
if input_data.reels_title:
facebook_options["reelsTitle"] = input_data.reels_title
if input_data.reels_thumbnail:
facebook_options["reelsThumbnail"] = input_data.reels_thumbnail
if input_data.is_story:
facebook_options["isStory"] = True
if input_data.media_captions:
facebook_options["mediaCaptions"] = input_data.media_captions
if input_data.location_id:
facebook_options["locationId"] = input_data.location_id
if input_data.age_min > 0:
facebook_options["ageMin"] = input_data.age_min
if input_data.target_countries:
facebook_options["targetCountries"] = input_data.target_countries
if input_data.alt_text:
facebook_options["altText"] = input_data.alt_text
if input_data.video_title:
facebook_options["videoTitle"] = input_data.video_title
if input_data.video_thumbnail:
facebook_options["videoThumbnail"] = input_data.video_thumbnail
if input_data.is_draft:
facebook_options["isDraft"] = True
if input_data.scheduled_publish_date:
facebook_options["scheduledPublishDate"] = input_data.scheduled_publish_date
if input_data.preview_link:
facebook_options["previewLink"] = input_data.preview_link
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.FACEBOOK],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
facebook_options=facebook_options if facebook_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

210
autogpt_platform/backend/backend/blocks/ayrshare/post_to_gmb.py Normal file
View File

@@ -0,0 +1,210 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToGMBBlock(Block):
"""Block for posting to Google My Business with GMB-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Google My Business posts."""
# Override media_urls to include GMB-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. GMB supports only one image or video per post.",
default_factory=list,
advanced=False,
)
# GMB-specific options
is_photo_video: bool = SchemaField(
description="Whether this is a photo/video post (appears in Photos section)",
default=False,
advanced=True,
)
photo_category: str = SchemaField(
description="Category for photo/video: cover, profile, logo, exterior, interior, product, at_work, food_and_drink, menu, common_area, rooms, teams",
default="",
advanced=True,
)
# Call to action options (flattened from CallToAction object)
call_to_action_type: str = SchemaField(
description="Type of action button: 'book', 'order', 'shop', 'learn_more', 'sign_up', or 'call'",
default="",
advanced=True,
)
call_to_action_url: str = SchemaField(
description="URL for the action button (not required for 'call' action)",
default="",
advanced=True,
)
# Event details options (flattened from EventDetails object)
event_title: str = SchemaField(
description="Event title for event posts",
default="",
advanced=True,
)
event_start_date: str = SchemaField(
description="Event start date in ISO format (e.g., '2024-03-15T09:00:00Z')",
default="",
advanced=True,
)
event_end_date: str = SchemaField(
description="Event end date in ISO format (e.g., '2024-03-15T17:00:00Z')",
default="",
advanced=True,
)
# Offer details options (flattened from OfferDetails object)
offer_title: str = SchemaField(
description="Offer title for promotional posts",
default="",
advanced=True,
)
offer_start_date: str = SchemaField(
description="Offer start date in ISO format (e.g., '2024-03-15T00:00:00Z')",
default="",
advanced=True,
)
offer_end_date: str = SchemaField(
description="Offer end date in ISO format (e.g., '2024-04-15T23:59:59Z')",
default="",
advanced=True,
)
offer_coupon_code: str = SchemaField(
description="Coupon code for the offer (max 58 characters)",
default="",
advanced=True,
)
offer_redeem_online_url: str = SchemaField(
description="URL where customers can redeem the offer online",
default="",
advanced=True,
)
offer_terms_conditions: str = SchemaField(
description="Terms and conditions for the offer",
default="",
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="2c38c783-c484-4503-9280-ef5d1d345a7e",
description="Post to Google My Business using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToGMBBlock.Input,
output_schema=PostToGMBBlock.Output,
)
async def run(
self, input_data: "PostToGMBBlock.Input", *, profile_key: SecretStr, **kwargs
) -> BlockOutput:
"""Post to Google My Business with GMB-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate GMB constraints
if len(input_data.media_urls) > 1:
yield "error", "Google My Business supports only one image or video per post"
return
# Validate offer coupon code length
if input_data.offer_coupon_code and len(input_data.offer_coupon_code) > 58:
yield "error", "GMB offer coupon code cannot exceed 58 characters"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build GMB-specific options
gmb_options = {}
# Photo/Video post options
if input_data.is_photo_video:
gmb_options["isPhotoVideo"] = True
if input_data.photo_category:
gmb_options["category"] = input_data.photo_category
# Call to Action (from flattened fields)
if input_data.call_to_action_type:
cta_dict = {"actionType": input_data.call_to_action_type}
# URL not required for 'call' action type
if (
input_data.call_to_action_type != "call"
and input_data.call_to_action_url
):
cta_dict["url"] = input_data.call_to_action_url
gmb_options["callToAction"] = cta_dict
# Event details (from flattened fields)
if (
input_data.event_title
and input_data.event_start_date
and input_data.event_end_date
):
gmb_options["event"] = {
"title": input_data.event_title,
"startDate": input_data.event_start_date,
"endDate": input_data.event_end_date,
}
# Offer details (from flattened fields)
if (
input_data.offer_title
and input_data.offer_start_date
and input_data.offer_end_date
and input_data.offer_coupon_code
and input_data.offer_redeem_online_url
and input_data.offer_terms_conditions
):
gmb_options["offer"] = {
"title": input_data.offer_title,
"startDate": input_data.offer_start_date,
"endDate": input_data.offer_end_date,
"couponCode": input_data.offer_coupon_code,
"redeemOnlineUrl": input_data.offer_redeem_online_url,
"termsConditions": input_data.offer_terms_conditions,
}
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.GOOGLE_MY_BUSINESS],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
gmb_options=gmb_options if gmb_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

221
autogpt_platform/backend/backend/blocks/ayrshare/post_to_instagram.py Normal file
View File

@@ -0,0 +1,221 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, InstagramUserTag, create_ayrshare_client
class PostToInstagramBlock(Block):
"""Block for posting to Instagram with Instagram-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Instagram posts."""
# Override post field to include Instagram-specific information
post: str = SchemaField(
description="The post text (max 2,200 chars, up to 30 hashtags, 3 @mentions)",
default="",
advanced=False,
)
# Override media_urls to include Instagram-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. Instagram supports up to 10 images/videos in a carousel.",
default_factory=list,
advanced=False,
)
# Instagram-specific options
is_story: bool = SchemaField(
description="Whether to post as Instagram Story (24-hour expiration)",
default=False,
advanced=True,
)
share_reels_feed: bool = SchemaField(
description="Whether Reel should appear in both Feed and Reels tabs",
default=True,
advanced=True,
)
audio_name: str = SchemaField(
description="Audio name for Reels (e.g., 'The Weeknd - Blinding Lights')",
default="",
advanced=True,
)
thumbnail: str = SchemaField(
description="Thumbnail URL for Reel video", default="", advanced=True
)
thumbnail_offset: int = SchemaField(
description="Thumbnail frame offset in milliseconds (default: 0)",
default=0,
advanced=True,
)
alt_text: list[str] = SchemaField(
description="Alt text for each media item (up to 1,000 chars each, accessibility feature)",
default_factory=list,
advanced=True,
)
location_id: str = SchemaField(
description="Facebook Page ID or name for location tagging (e.g., '7640348500' or '@guggenheimmuseum')",
default="",
advanced=True,
)
user_tags: list[InstagramUserTag] = SchemaField(
description="List of users to tag with coordinates for images",
default_factory=list,
advanced=True,
)
collaborators: list[str] = SchemaField(
description="Instagram usernames to invite as collaborators (max 3, public accounts only)",
default_factory=list,
advanced=True,
)
auto_resize: bool = SchemaField(
description="Auto-resize images to 1080x1080px for Instagram",
default=False,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="89b02b96-a7cb-46f4-9900-c48b32fe1552",
description="Post to Instagram using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToInstagramBlock.Input,
output_schema=PostToInstagramBlock.Output,
)
async def run(
self,
input_data: "PostToInstagramBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Instagram with Instagram-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate Instagram constraints
if len(input_data.post) > 2200:
yield "error", f"Instagram post text exceeds 2,200 character limit ({len(input_data.post)} characters)"
return
if len(input_data.media_urls) > 10:
yield "error", "Instagram supports a maximum of 10 images/videos in a carousel"
return
if len(input_data.collaborators) > 3:
yield "error", "Instagram supports a maximum of 3 collaborators"
return
# Count hashtags and mentions
hashtag_count = input_data.post.count("#")
mention_count = input_data.post.count("@")
if hashtag_count > 30:
yield "error", f"Instagram allows maximum 30 hashtags ({hashtag_count} found)"
return
if mention_count > 3:
yield "error", f"Instagram allows maximum 3 @mentions ({mention_count} found)"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Instagram-specific options
instagram_options = {}
# Stories
if input_data.is_story:
instagram_options["stories"] = True
# Reels options
if input_data.share_reels_feed is not None:
instagram_options["shareReelsFeed"] = input_data.share_reels_feed
if input_data.audio_name:
instagram_options["audioName"] = input_data.audio_name
if input_data.thumbnail:
instagram_options["thumbNail"] = input_data.thumbnail
elif input_data.thumbnail_offset > 0:
instagram_options["thumbNailOffset"] = input_data.thumbnail_offset
# Alt text
if input_data.alt_text:
# Validate alt text length
for i, alt in enumerate(input_data.alt_text):
if len(alt) > 1000:
yield "error", f"Alt text {i+1} exceeds 1,000 character limit ({len(alt)} characters)"
return
instagram_options["altText"] = input_data.alt_text
# Location
if input_data.location_id:
instagram_options["locationId"] = input_data.location_id
# User tags
if input_data.user_tags:
user_tags_list = []
for tag in input_data.user_tags:
tag_dict: dict[str, float | str] = {"username": tag.username}
if tag.x is not None and tag.y is not None:
# Validate coordinates
if not (0.0 <= tag.x <= 1.0) or not (0.0 <= tag.y <= 1.0):
yield "error", f"User tag coordinates must be between 0.0 and 1.0 (user: {tag.username})"
return
tag_dict["x"] = tag.x
tag_dict["y"] = tag.y
user_tags_list.append(tag_dict)
instagram_options["userTags"] = user_tags_list
# Collaborators
if input_data.collaborators:
instagram_options["collaborators"] = input_data.collaborators
# Auto resize
if input_data.auto_resize:
instagram_options["autoResize"] = True
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.INSTAGRAM],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
instagram_options=instagram_options if instagram_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

222
autogpt_platform/backend/backend/blocks/ayrshare/post_to_linkedin.py Normal file
View File

@@ -0,0 +1,222 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToLinkedInBlock(Block):
"""Block for posting to LinkedIn with LinkedIn-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for LinkedIn posts."""
# Override post field to include LinkedIn-specific information
post: str = SchemaField(
description="The post text (max 3,000 chars, hashtags supported with #)",
default="",
advanced=False,
)
# Override media_urls to include LinkedIn-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. LinkedIn supports up to 9 images, videos, or documents (PPT, PPTX, DOC, DOCX, PDF <100MB, <300 pages).",
default_factory=list,
advanced=False,
)
# LinkedIn-specific options
visibility: str = SchemaField(
description="Post visibility: 'public' (default), 'connections' (personal only), 'loggedin'",
default="public",
advanced=True,
)
alt_text: list[str] = SchemaField(
description="Alt text for each image (accessibility feature, not supported for videos/documents)",
default_factory=list,
advanced=True,
)
titles: list[str] = SchemaField(
description="Title/caption for each image or video",
default_factory=list,
advanced=True,
)
document_title: str = SchemaField(
description="Title for document posts (max 400 chars, uses filename if not specified)",
default="",
advanced=True,
)
thumbnail: str = SchemaField(
description="Thumbnail URL for video (PNG/JPG, same dimensions as video, <10MB)",
default="",
advanced=True,
)
# LinkedIn targeting options (flattened from LinkedInTargeting object)
targeting_countries: list[str] | None = SchemaField(
description="Country codes for targeting (e.g., ['US', 'IN', 'DE', 'GB']). Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_seniorities: list[str] | None = SchemaField(
description="Seniority levels for targeting (e.g., ['Senior', 'VP']). Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_degrees: list[str] | None = SchemaField(
description="Education degrees for targeting. Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_fields_of_study: list[str] | None = SchemaField(
description="Fields of study for targeting. Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_industries: list[str] | None = SchemaField(
description="Industry categories for targeting. Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_job_functions: list[str] | None = SchemaField(
description="Job function categories for targeting. Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
targeting_staff_count_ranges: list[str] | None = SchemaField(
description="Company size ranges for targeting. Requires 300+ followers in target audience.",
default=None,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
id="589af4e4-507f-42fd-b9ac-a67ecef25811",
description="Post to LinkedIn using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToLinkedInBlock.Input,
output_schema=PostToLinkedInBlock.Output,
)
async def run(
self,
input_data: "PostToLinkedInBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to LinkedIn with LinkedIn-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate LinkedIn constraints
if len(input_data.post) > 3000:
yield "error", f"LinkedIn post text exceeds 3,000 character limit ({len(input_data.post)} characters)"
return
if len(input_data.media_urls) > 9:
yield "error", "LinkedIn supports a maximum of 9 images/videos/documents"
return
if input_data.document_title and len(input_data.document_title) > 400:
yield "error", f"LinkedIn document title exceeds 400 character limit ({len(input_data.document_title)} characters)"
return
# Validate visibility option
valid_visibility = ["public", "connections", "loggedin"]
if input_data.visibility not in valid_visibility:
yield "error", f"LinkedIn visibility must be one of: {', '.join(valid_visibility)}"
return
# Check for document extensions
document_extensions = [".ppt", ".pptx", ".doc", ".docx", ".pdf"]
has_documents = any(
any(url.lower().endswith(ext) for ext in document_extensions)
for url in input_data.media_urls
)
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build LinkedIn-specific options
linkedin_options = {}
# Visibility
if input_data.visibility != "public":
linkedin_options["visibility"] = input_data.visibility
# Alt text (not supported for videos or documents)
if input_data.alt_text and not has_documents:
linkedin_options["altText"] = input_data.alt_text
# Titles/captions
if input_data.titles:
linkedin_options["titles"] = input_data.titles
# Document title
if input_data.document_title and has_documents:
linkedin_options["title"] = input_data.document_title
# Video thumbnail
if input_data.thumbnail:
linkedin_options["thumbNail"] = input_data.thumbnail
# Audience targeting (from flattened fields)
targeting_dict = {}
if input_data.targeting_countries:
targeting_dict["countries"] = input_data.targeting_countries
if input_data.targeting_seniorities:
targeting_dict["seniorities"] = input_data.targeting_seniorities
if input_data.targeting_degrees:
targeting_dict["degrees"] = input_data.targeting_degrees
if input_data.targeting_fields_of_study:
targeting_dict["fieldsOfStudy"] = input_data.targeting_fields_of_study
if input_data.targeting_industries:
targeting_dict["industries"] = input_data.targeting_industries
if input_data.targeting_job_functions:
targeting_dict["jobFunctions"] = input_data.targeting_job_functions
if input_data.targeting_staff_count_ranges:
targeting_dict["staffCountRanges"] = input_data.targeting_staff_count_ranges
if targeting_dict:
linkedin_options["targeting"] = targeting_dict
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.LINKEDIN],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
linkedin_options=linkedin_options if linkedin_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

209
autogpt_platform/backend/backend/blocks/ayrshare/post_to_pinterest.py Normal file
View File

@@ -0,0 +1,209 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, PinterestCarouselOption, create_ayrshare_client
class PostToPinterestBlock(Block):
"""Block for posting to Pinterest with Pinterest-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Pinterest posts."""
# Override post field to include Pinterest-specific information
post: str = SchemaField(
description="Pin description (max 500 chars, links not clickable - use link field instead)",
default="",
advanced=False,
)
# Override media_urls to include Pinterest-specific constraints
media_urls: list[str] = SchemaField(
description="Required image/video URLs. Pinterest requires at least one image. Videos need thumbnail. Up to 5 images for carousel.",
default_factory=list,
advanced=False,
)
# Pinterest-specific options
pin_title: str = SchemaField(
description="Pin title displayed in 'Add your title' section (max 100 chars)",
default="",
advanced=True,
)
link: str = SchemaField(
description="Clickable destination URL when users click the pin (max 2048 chars)",
default="",
advanced=True,
)
board_id: str = SchemaField(
description="Pinterest Board ID to post to (from /user/details endpoint, uses default board if not specified)",
default="",
advanced=True,
)
note: str = SchemaField(
description="Private note for the pin (only visible to you and board collaborators)",
default="",
advanced=True,
)
thumbnail: str = SchemaField(
description="Required thumbnail URL for video pins (must have valid image Content-Type)",
default="",
advanced=True,
)
carousel_options: list[PinterestCarouselOption] = SchemaField(
description="Options for each image in carousel (title, link, description per image)",
default_factory=list,
advanced=True,
)
alt_text: list[str] = SchemaField(
description="Alt text for each image/video (max 500 chars each, accessibility feature)",
default_factory=list,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="3ca46e05-dbaa-4afb-9e95-5a429c4177e6",
description="Post to Pinterest using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToPinterestBlock.Input,
output_schema=PostToPinterestBlock.Output,
)
async def run(
self,
input_data: "PostToPinterestBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Pinterest with Pinterest-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate Pinterest constraints
if len(input_data.post) > 500:
yield "error", f"Pinterest pin description exceeds 500 character limit ({len(input_data.post)} characters)"
return
if len(input_data.pin_title) > 100:
yield "error", f"Pinterest pin title exceeds 100 character limit ({len(input_data.pin_title)} characters)"
return
if len(input_data.link) > 2048:
yield "error", f"Pinterest link URL exceeds 2048 character limit ({len(input_data.link)} characters)"
return
if len(input_data.media_urls) == 0:
yield "error", "Pinterest requires at least one image or video"
return
if len(input_data.media_urls) > 5:
yield "error", "Pinterest supports a maximum of 5 images in a carousel"
return
# Check if video is included and thumbnail is provided
video_extensions = [".mp4", ".mov", ".avi", ".mkv", ".wmv", ".flv", ".webm"]
has_video = any(
any(url.lower().endswith(ext) for ext in video_extensions)
for url in input_data.media_urls
)
if (has_video or input_data.is_video) and not input_data.thumbnail:
yield "error", "Pinterest video pins require a thumbnail URL"
return
# Validate alt text length
for i, alt in enumerate(input_data.alt_text):
if len(alt) > 500:
yield "error", f"Pinterest alt text {i+1} exceeds 500 character limit ({len(alt)} characters)"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Pinterest-specific options
pinterest_options = {}
# Pin title
if input_data.pin_title:
pinterest_options["title"] = input_data.pin_title
# Clickable link
if input_data.link:
pinterest_options["link"] = input_data.link
# Board ID
if input_data.board_id:
pinterest_options["boardId"] = input_data.board_id
# Private note
if input_data.note:
pinterest_options["note"] = input_data.note
# Video thumbnail
if input_data.thumbnail:
pinterest_options["thumbNail"] = input_data.thumbnail
# Carousel options
if input_data.carousel_options:
carousel_list = []
for option in input_data.carousel_options:
carousel_dict = {}
if option.title:
carousel_dict["title"] = option.title
if option.link:
carousel_dict["link"] = option.link
if option.description:
carousel_dict["description"] = option.description
if carousel_dict: # Only add if not empty
carousel_list.append(carousel_dict)
if carousel_list:
pinterest_options["carouselOptions"] = carousel_list
# Alt text
if input_data.alt_text:
pinterest_options["altText"] = input_data.alt_text
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.PINTEREST],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
pinterest_options=pinterest_options if pinterest_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

69
autogpt_platform/backend/backend/blocks/ayrshare/post_to_reddit.py Normal file
View File

@@ -0,0 +1,69 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToRedditBlock(Block):
"""Block for posting to Reddit."""
class Input(BaseAyrshareInput):
"""Input schema for Reddit posts."""
pass # Uses all base fields
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="c7733580-3c72-483e-8e47-a8d58754d853",
description="Post to Reddit using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToRedditBlock.Input,
output_schema=PostToRedditBlock.Output,
)
async def run(
self, input_data: "PostToRedditBlock.Input", *, profile_key: SecretStr, **kwargs
) -> BlockOutput:
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured."
return
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.REDDIT],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

129
autogpt_platform/backend/backend/blocks/ayrshare/post_to_snapchat.py Normal file
View File

@@ -0,0 +1,129 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToSnapchatBlock(Block):
"""Block for posting to Snapchat with Snapchat-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Snapchat posts."""
# Override post field to include Snapchat-specific information
post: str = SchemaField(
description="The post text (optional for video-only content)",
default="",
advanced=False,
)
# Override media_urls to include Snapchat-specific constraints
media_urls: list[str] = SchemaField(
description="Required video URL for Snapchat posts. Snapchat only supports video content.",
default_factory=list,
advanced=False,
)
# Snapchat-specific options
story_type: str = SchemaField(
description="Type of Snapchat content: 'story' (24-hour Stories), 'saved_story' (Saved Stories), or 'spotlight' (Spotlight posts)",
default="story",
advanced=True,
)
video_thumbnail: str = SchemaField(
description="Thumbnail URL for video content (optional, auto-generated if not provided)",
default="",
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="a9d7f854-2c83-4e96-b3a1-7f2e9c5d4b8e",
description="Post to Snapchat using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToSnapchatBlock.Input,
output_schema=PostToSnapchatBlock.Output,
)
async def run(
self,
input_data: "PostToSnapchatBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Snapchat with Snapchat-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate Snapchat constraints
if not input_data.media_urls:
yield "error", "Snapchat requires at least one video URL"
return
if len(input_data.media_urls) > 1:
yield "error", "Snapchat supports only one video per post"
return
# Validate story type
valid_story_types = ["story", "saved_story", "spotlight"]
if input_data.story_type not in valid_story_types:
yield "error", f"Snapchat story type must be one of: {', '.join(valid_story_types)}"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Snapchat-specific options
snapchat_options = {}
# Story type
if input_data.story_type != "story":
snapchat_options["storyType"] = input_data.story_type
# Video thumbnail
if input_data.video_thumbnail:
snapchat_options["videoThumbnail"] = input_data.video_thumbnail
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.SNAPCHAT],
media_urls=input_data.media_urls,
is_video=True, # Snapchat only supports video
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
snapchat_options=snapchat_options if snapchat_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

116
autogpt_platform/backend/backend/blocks/ayrshare/post_to_telegram.py Normal file
View File

@@ -0,0 +1,116 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToTelegramBlock(Block):
"""Block for posting to Telegram with Telegram-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Telegram posts."""
# Override post field to include Telegram-specific information
post: str = SchemaField(
description="The post text (empty string allowed). Use @handle to mention other Telegram users.",
default="",
advanced=False,
)
# Override media_urls to include Telegram-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. For animated GIFs, only one URL is allowed. Telegram will auto-preview links unless image/video is included.",
default_factory=list,
advanced=False,
)
# Override is_video to include GIF-specific information
is_video: bool = SchemaField(
description="Whether the media is a video. Set to true for animated GIFs that don't end in .gif/.GIF extension.",
default=False,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="47bc74eb-4af2-452c-b933-af377c7287df",
description="Post to Telegram using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToTelegramBlock.Input,
output_schema=PostToTelegramBlock.Output,
)
async def run(
self,
input_data: "PostToTelegramBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Telegram with Telegram-specific validation."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate Telegram constraints
# Check for animated GIFs - only one URL allowed
gif_extensions = [".gif", ".GIF"]
has_gif = any(
any(url.endswith(ext) for ext in gif_extensions)
for url in input_data.media_urls
)
if has_gif and len(input_data.media_urls) > 1:
yield "error", "Telegram animated GIFs support only one URL per post"
return
# Auto-detect if we need to set is_video for GIFs without proper extension
detected_is_video = input_data.is_video
if input_data.media_urls and not has_gif and not input_data.is_video:
# Check if this might be a GIF without proper extension
# This is just informational - user needs to set is_video manually
pass
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.TELEGRAM],
media_urls=input_data.media_urls,
is_video=detected_is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

111
autogpt_platform/backend/backend/blocks/ayrshare/post_to_threads.py Normal file
View File

@@ -0,0 +1,111 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToThreadsBlock(Block):
"""Block for posting to Threads with Threads-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for Threads posts."""
# Override post field to include Threads-specific information
post: str = SchemaField(
description="The post text (max 500 chars, empty string allowed). Only 1 hashtag allowed. Use @handle to mention users.",
default="",
advanced=False,
)
# Override media_urls to include Threads-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. Supports up to 20 images/videos in a carousel. Auto-preview links unless media is included.",
default_factory=list,
advanced=False,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="f8c3b2e1-9d4a-4e5f-8c7b-6a9e8d2f1c3b",
description="Post to Threads using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToThreadsBlock.Input,
output_schema=PostToThreadsBlock.Output,
)
async def run(
self,
input_data: "PostToThreadsBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to Threads with Threads-specific validation."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate Threads constraints
if len(input_data.post) > 500:
yield "error", f"Threads post text exceeds 500 character limit ({len(input_data.post)} characters)"
return
if len(input_data.media_urls) > 20:
yield "error", "Threads supports a maximum of 20 images/videos in a carousel"
return
# Count hashtags (only 1 allowed)
hashtag_count = input_data.post.count("#")
if hashtag_count > 1:
yield "error", f"Threads allows only 1 hashtag per post ({hashtag_count} found)"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build Threads-specific options
threads_options = {}
# Note: Based on the documentation, Threads doesn't seem to have specific options
# beyond the standard ones. The main constraints are validation-based.
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.THREADS],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
threads_options=threads_options if threads_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

243
autogpt_platform/backend/backend/blocks/ayrshare/post_to_tiktok.py Normal file
View File

@@ -0,0 +1,243 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToTikTokBlock(Block):
"""Block for posting to TikTok with TikTok-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for TikTok posts."""
# Override post field to include TikTok-specific information
post: str = SchemaField(
description="The post text (max 2,200 chars, empty string allowed). Use @handle to mention users. Line breaks will be ignored.",
default="",
advanced=False,
)
# Override media_urls to include TikTok-specific constraints
media_urls: list[str] = SchemaField(
description="Required media URLs. Either 1 video OR up to 35 images (JPG/JPEG/WEBP only). Cannot mix video and images.",
default_factory=list,
advanced=False,
)
# TikTok-specific options
auto_add_music: bool = SchemaField(
description="Automatically add recommended music to image posts",
default=False,
advanced=True,
)
disable_comments: bool = SchemaField(
description="Disable comments on the published post",
default=False,
advanced=True,
)
disable_duet: bool = SchemaField(
description="Disable duets on published video (video only)",
default=False,
advanced=True,
)
disable_stitch: bool = SchemaField(
description="Disable stitch on published video (video only)",
default=False,
advanced=True,
)
is_ai_generated: bool = SchemaField(
description="Label content as AI-generated (video only)",
default=False,
advanced=True,
)
is_branded_content: bool = SchemaField(
description="Label as branded content (paid partnership)",
default=False,
advanced=True,
)
is_brand_organic: bool = SchemaField(
description="Label as brand organic content (promotional)",
default=False,
advanced=True,
)
image_cover_index: int = SchemaField(
description="Index of image to use as cover (0-based, image posts only)",
default=0,
advanced=True,
)
title: str = SchemaField(
description="Title for image posts", default="", advanced=True
)
thumbnail_offset: int = SchemaField(
description="Video thumbnail frame offset in milliseconds (video only)",
default=0,
advanced=True,
)
visibility: str = SchemaField(
description="Post visibility: 'public', 'private', 'followers', or 'friends'",
default="public",
advanced=True,
)
draft: bool = SchemaField(
description="Create as draft post (video only)",
default=False,
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="7faf4b27-96b0-4f05-bf64-e0de54ae74e1",
description="Post to TikTok using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToTikTokBlock.Input,
output_schema=PostToTikTokBlock.Output,
)
async def run(
self, input_data: "PostToTikTokBlock.Input", *, profile_key: SecretStr, **kwargs
) -> BlockOutput:
"""Post to TikTok with TikTok-specific validation and options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate TikTok constraints
if len(input_data.post) > 2200:
yield "error", f"TikTok post text exceeds 2,200 character limit ({len(input_data.post)} characters)"
return
if not input_data.media_urls:
yield "error", "TikTok requires at least one media URL (either 1 video or up to 35 images)"
return
# Check for video vs image constraints
video_extensions = [".mp4", ".mov", ".avi", ".mkv", ".wmv", ".flv", ".webm"]
image_extensions = [".jpg", ".jpeg", ".webp"]
has_video = input_data.is_video or any(
any(url.lower().endswith(ext) for ext in video_extensions)
for url in input_data.media_urls
)
has_images = any(
any(url.lower().endswith(ext) for ext in image_extensions)
for url in input_data.media_urls
)
if has_video and has_images:
yield "error", "TikTok does not support mixing video and images in the same post"
return
if has_video and len(input_data.media_urls) > 1:
yield "error", "TikTok supports only 1 video per post"
return
if has_images and len(input_data.media_urls) > 35:
yield "error", "TikTok supports a maximum of 35 images per post"
return
# Validate image cover index
if has_images and input_data.image_cover_index >= len(input_data.media_urls):
yield "error", f"Image cover index {input_data.image_cover_index} is out of range (max: {len(input_data.media_urls) - 1})"
return
# Validate visibility option
valid_visibility = ["public", "private", "followers", "friends"]
if input_data.visibility not in valid_visibility:
yield "error", f"TikTok visibility must be one of: {', '.join(valid_visibility)}"
return
# Check for PNG files (not supported)
has_png = any(url.lower().endswith(".png") for url in input_data.media_urls)
if has_png:
yield "error", "TikTok does not support PNG files. Please use JPG, JPEG, or WEBP for images."
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build TikTok-specific options
tiktok_options = {}
# Common options
if input_data.auto_add_music and has_images:
tiktok_options["autoAddMusic"] = True
if input_data.disable_comments:
tiktok_options["disableComments"] = True
if input_data.is_branded_content:
tiktok_options["isBrandedContent"] = True
if input_data.is_brand_organic:
tiktok_options["isBrandOrganic"] = True
# Video-specific options
if has_video:
if input_data.disable_duet:
tiktok_options["disableDuet"] = True
if input_data.disable_stitch:
tiktok_options["disableStitch"] = True
if input_data.is_ai_generated:
tiktok_options["isAIGenerated"] = True
if input_data.thumbnail_offset > 0:
tiktok_options["thumbNailOffset"] = input_data.thumbnail_offset
if input_data.draft:
tiktok_options["draft"] = True
# Image-specific options
if has_images:
if input_data.image_cover_index > 0:
tiktok_options["imageCoverIndex"] = input_data.image_cover_index
if input_data.title:
tiktok_options["title"] = input_data.title
if input_data.visibility != "public":
tiktok_options["visibility"] = input_data.visibility
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.TIKTOK],
media_urls=input_data.media_urls,
is_video=has_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
tiktok_options=tiktok_options if tiktok_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

241
autogpt_platform/backend/backend/blocks/ayrshare/post_to_x.py Normal file
View File

@@ -0,0 +1,241 @@
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToXBlock(Block):
"""Block for posting to X / Twitter with Twitter-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for X / Twitter posts."""
# Override post field to include X-specific information
post: str = SchemaField(
description="The post text (max 280 chars, up to 25,000 for Premium users). Use @handle to mention users. Use \\n\\n for thread breaks.",
advanced=False,
)
# Override media_urls to include X-specific constraints
media_urls: list[str] = SchemaField(
description="Optional list of media URLs. X supports up to 4 images or videos per tweet. Auto-preview links unless media is included.",
default_factory=list,
advanced=False,
)
# X-specific options
reply_to_id: str | None = SchemaField(
description="ID of the tweet to reply to",
default=None,
advanced=True,
)
quote_tweet_id: str | None = SchemaField(
description="ID of the tweet to quote (low-level Tweet ID)",
default=None,
advanced=True,
)
poll_options: list[str] = SchemaField(
description="Poll options (2-4 choices)",
default_factory=list,
advanced=True,
)
poll_duration: int = SchemaField(
description="Poll duration in minutes (1-10080)",
default=1440,
advanced=True,
)
alt_text: list[str] = SchemaField(
description="Alt text for each image (max 1,000 chars each, not supported for videos)",
default_factory=list,
advanced=True,
)
is_thread: bool = SchemaField(
description="Whether to automatically break post into thread based on line breaks",
default=False,
advanced=True,
)
thread_number: bool = SchemaField(
description="Add thread numbers (1/n format) to each thread post",
default=False,
advanced=True,
)
thread_media_urls: list[str] = SchemaField(
description="Media URLs for thread posts (one per thread, use 'null' to skip)",
default_factory=list,
advanced=True,
)
long_post: bool = SchemaField(
description="Force long form post (requires Premium X account)",
default=False,
advanced=True,
)
long_video: bool = SchemaField(
description="Enable long video upload (requires approval and Business/Enterprise plan)",
default=False,
advanced=True,
)
subtitle_url: str = SchemaField(
description="URL to SRT subtitle file for videos (must be HTTPS and end in .srt)",
default="",
advanced=True,
)
subtitle_language: str = SchemaField(
description="Language code for subtitles (default: 'en')",
default="en",
advanced=True,
)
subtitle_name: str = SchemaField(
description="Name of caption track (max 150 chars, default: 'English')",
default="English",
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
id="9e8f844e-b4a5-4b25-80f2-9e1dd7d67625",
description="Post to X / Twitter using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToXBlock.Input,
output_schema=PostToXBlock.Output,
)
async def run(
self,
input_data: "PostToXBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to X / Twitter with enhanced X-specific options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate X constraints
if not input_data.long_post and len(input_data.post) > 280:
yield "error", f"X post text exceeds 280 character limit ({len(input_data.post)} characters). Enable 'long_post' for Premium accounts."
return
if input_data.long_post and len(input_data.post) > 25000:
yield "error", f"X long post text exceeds 25,000 character limit ({len(input_data.post)} characters)"
return
if len(input_data.media_urls) > 4:
yield "error", "X supports a maximum of 4 images or videos per tweet"
return
# Validate poll options
if input_data.poll_options:
if len(input_data.poll_options) < 2 or len(input_data.poll_options) > 4:
yield "error", "X polls require 2-4 options"
return
if input_data.poll_duration < 1 or input_data.poll_duration > 10080:
yield "error", "X poll duration must be between 1 and 10,080 minutes (7 days)"
return
# Validate alt text
if input_data.alt_text:
for i, alt in enumerate(input_data.alt_text):
if len(alt) > 1000:
yield "error", f"X alt text {i+1} exceeds 1,000 character limit ({len(alt)} characters)"
return
# Validate subtitle settings
if input_data.subtitle_url:
if not input_data.subtitle_url.startswith(
"https://"
) or not input_data.subtitle_url.endswith(".srt"):
yield "error", "Subtitle URL must start with https:// and end with .srt"
return
if len(input_data.subtitle_name) > 150:
yield "error", f"Subtitle name exceeds 150 character limit ({len(input_data.subtitle_name)} characters)"
return
# Convert datetime to ISO format if provided
iso_date = (
input_data.schedule_date.isoformat() if input_data.schedule_date else None
)
# Build X-specific options
twitter_options = {}
# Basic options
if input_data.reply_to_id:
twitter_options["replyToId"] = input_data.reply_to_id
if input_data.quote_tweet_id:
twitter_options["quoteTweetId"] = input_data.quote_tweet_id
if input_data.long_post:
twitter_options["longPost"] = True
if input_data.long_video:
twitter_options["longVideo"] = True
# Poll options
if input_data.poll_options:
twitter_options["poll"] = {
"duration": input_data.poll_duration,
"options": input_data.poll_options,
}
# Alt text for images
if input_data.alt_text:
twitter_options["altText"] = input_data.alt_text
# Thread options
if input_data.is_thread:
twitter_options["thread"] = True
if input_data.thread_number:
twitter_options["threadNumber"] = True
if input_data.thread_media_urls:
twitter_options["mediaUrls"] = input_data.thread_media_urls
# Video subtitle options
if input_data.subtitle_url:
twitter_options["subTitleUrl"] = input_data.subtitle_url
twitter_options["subTitleLanguage"] = input_data.subtitle_language
twitter_options["subTitleName"] = input_data.subtitle_name
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.TWITTER],
media_urls=input_data.media_urls,
is_video=input_data.is_video,
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
twitter_options=twitter_options if twitter_options else None,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

305
autogpt_platform/backend/backend/blocks/ayrshare/post_to_youtube.py Normal file
View File

@@ -0,0 +1,305 @@
from typing import Any
from backend.integrations.ayrshare import PostIds, PostResponse, SocialPlatform
from backend.sdk import (
Block,
BlockCategory,
BlockOutput,
BlockSchema,
BlockType,
SchemaField,
SecretStr,
)
from ._util import BaseAyrshareInput, create_ayrshare_client
class PostToYouTubeBlock(Block):
"""Block for posting to YouTube with YouTube-specific options."""
class Input(BaseAyrshareInput):
"""Input schema for YouTube posts."""
# Override post field to include YouTube-specific information
post: str = SchemaField(
description="Video description (max 5,000 chars, empty string allowed). Cannot contain < or > characters.",
default="",
advanced=False,
)
# Override media_urls to include YouTube-specific constraints
media_urls: list[str] = SchemaField(
description="Required video URL. YouTube only supports 1 video per post.",
default_factory=list,
advanced=False,
)
# YouTube-specific required options
title: str = SchemaField(
description="Video title (max 100 chars, required). Cannot contain < or > characters.",
default="",
advanced=False,
)
# YouTube-specific optional options
visibility: str = SchemaField(
description="Video visibility: 'private' (default), 'public', or 'unlisted'",
default="private",
advanced=True,
)
thumbnail: str = SchemaField(
description="Thumbnail URL (JPEG/PNG under 2MB, must end in .png/.jpg/.jpeg). Requires phone verification.",
default="",
advanced=True,
)
playlist_id: str = SchemaField(
description="Playlist ID to add video (user must own playlist)",
default="",
advanced=True,
)
tags: list[str] = SchemaField(
description="Video tags (min 2 chars each, max 500 chars total)",
default_factory=list,
advanced=True,
)
made_for_kids: bool = SchemaField(
description="Self-declared kids content", default=False, advanced=True
)
is_shorts: bool = SchemaField(
description="Post as YouTube Short (max 3 minutes, adds #shorts)",
default=False,
advanced=True,
)
notify_subscribers: bool = SchemaField(
description="Send notification to subscribers", default=True, advanced=True
)
category_id: int = SchemaField(
description="Video category ID (e.g., 24 = Entertainment)",
default=0,
advanced=True,
)
contains_synthetic_media: bool = SchemaField(
description="Disclose realistic AI/synthetic content",
default=False,
advanced=True,
)
publish_at: str = SchemaField(
description="UTC publish time (YouTube controlled, format: 2022-10-08T21:18:36Z)",
default="",
advanced=True,
)
# YouTube targeting options (flattened from YouTubeTargeting object)
targeting_block_countries: list[str] | None = SchemaField(
description="Country codes to block from viewing (e.g., ['US', 'CA'])",
default=None,
advanced=True,
)
targeting_allow_countries: list[str] | None = SchemaField(
description="Country codes to allow viewing (e.g., ['GB', 'AU'])",
default=None,
advanced=True,
)
subtitle_url: str = SchemaField(
description="URL to SRT or SBV subtitle file (must be HTTPS and end in .srt/.sbv, under 100MB)",
default="",
advanced=True,
)
subtitle_language: str = SchemaField(
description="Language code for subtitles (default: 'en')",
default="en",
advanced=True,
)
subtitle_name: str = SchemaField(
description="Name of caption track (max 150 chars, default: 'English')",
default="English",
advanced=True,
)
class Output(BlockSchema):
post_result: PostResponse = SchemaField(description="The result of the post")
post: PostIds = SchemaField(description="The result of the post")
def __init__(self):
super().__init__(
disabled=True,
id="0082d712-ff1b-4c3d-8a8d-6c7721883b83",
description="Post to YouTube using Ayrshare",
categories={BlockCategory.SOCIAL},
block_type=BlockType.AYRSHARE,
input_schema=PostToYouTubeBlock.Input,
output_schema=PostToYouTubeBlock.Output,
)
async def run(
self,
input_data: "PostToYouTubeBlock.Input",
*,
profile_key: SecretStr,
**kwargs,
) -> BlockOutput:
"""Post to YouTube with YouTube-specific validation and options."""
if not profile_key:
yield "error", "Please link a social account via Ayrshare"
return
client = create_ayrshare_client()
if not client:
yield "error", "Ayrshare integration is not configured. Please set up the AYRSHARE_API_KEY."
return
# Validate YouTube constraints
if not input_data.title:
yield "error", "YouTube requires a video title"
return
if len(input_data.title) > 100:
yield "error", f"YouTube title exceeds 100 character limit ({len(input_data.title)} characters)"
return
if len(input_data.post) > 5000:
yield "error", f"YouTube description exceeds 5,000 character limit ({len(input_data.post)} characters)"
return
# Check for forbidden characters
forbidden_chars = ["<", ">"]
for char in forbidden_chars:
if char in input_data.title:
yield "error", f"YouTube title cannot contain '{char}' character"
return
if char in input_data.post:
yield "error", f"YouTube description cannot contain '{char}' character"
return
if not input_data.media_urls:
yield "error", "YouTube requires exactly one video URL"
return
if len(input_data.media_urls) > 1:
yield "error", "YouTube supports only 1 video per post"
return
# Validate visibility option
valid_visibility = ["private", "public", "unlisted"]
if input_data.visibility not in valid_visibility:
yield "error", f"YouTube visibility must be one of: {', '.join(valid_visibility)}"
return
# Validate thumbnail URL format
if input_data.thumbnail:
valid_extensions = [".png", ".jpg", ".jpeg"]
if not any(
input_data.thumbnail.lower().endswith(ext) for ext in valid_extensions
):
yield "error", "YouTube thumbnail must end in .png, .jpg, or .jpeg"
return
# Validate tags
if input_data.tags:
total_tag_length = sum(len(tag) for tag in input_data.tags)
if total_tag_length > 500:
yield "error", f"YouTube tags total length exceeds 500 characters ({total_tag_length} characters)"
return
for tag in input_data.tags:
if len(tag) < 2:
yield "error", f"YouTube tag '{tag}' is too short (minimum 2 characters)"
return
# Validate subtitle URL
if input_data.subtitle_url:
if not input_data.subtitle_url.startswith("https://"):
yield "error", "YouTube subtitle URL must start with https://"
return
valid_subtitle_extensions = [".srt", ".sbv"]
if not any(
input_data.subtitle_url.lower().endswith(ext)
for ext in valid_subtitle_extensions
):
yield "error", "YouTube subtitle URL must end in .srt or .sbv"
return
if len(input_data.subtitle_name) > 150:
yield "error", f"YouTube subtitle name exceeds 150 character limit ({len(input_data.subtitle_name)} characters)"
return
# Validate publish_at format if provided
if input_data.publish_at and input_data.schedule_date:
yield "error", "Cannot use both 'publish_at' and 'schedule_date'. Use 'publish_at' for YouTube-controlled publishing."
return
# Convert datetime to ISO format if provided (only if not using publish_at)
iso_date = None
if not input_data.publish_at and input_data.schedule_date:
iso_date = input_data.schedule_date.isoformat()
# Build YouTube-specific options
youtube_options: dict[str, Any] = {"title": input_data.title}
# Basic options
if input_data.visibility != "private":
youtube_options["visibility"] = input_data.visibility
if input_data.thumbnail:
youtube_options["thumbNail"] = input_data.thumbnail
if input_data.playlist_id:
youtube_options["playListId"] = input_data.playlist_id
if input_data.tags:
youtube_options["tags"] = input_data.tags
if input_data.made_for_kids:
youtube_options["madeForKids"] = True
if input_data.is_shorts:
youtube_options["shorts"] = True
if not input_data.notify_subscribers:
youtube_options["notifySubscribers"] = False
if input_data.category_id > 0:
youtube_options["categoryId"] = input_data.category_id
if input_data.contains_synthetic_media:
youtube_options["containsSyntheticMedia"] = True
if input_data.publish_at:
youtube_options["publishAt"] = input_data.publish_at
# Country targeting (from flattened fields)
targeting_dict = {}
if input_data.targeting_block_countries:
targeting_dict["block"] = input_data.targeting_block_countries
if input_data.targeting_allow_countries:
targeting_dict["allow"] = input_data.targeting_allow_countries
if targeting_dict:
youtube_options["targeting"] = targeting_dict
# Subtitle options
if input_data.subtitle_url:
youtube_options["subTitleUrl"] = input_data.subtitle_url
youtube_options["subTitleLanguage"] = input_data.subtitle_language
youtube_options["subTitleName"] = input_data.subtitle_name
response = await client.create_post(
post=input_data.post,
platforms=[SocialPlatform.YOUTUBE],
media_urls=input_data.media_urls,
is_video=True, # YouTube only supports videos
schedule_date=iso_date,
disable_comments=input_data.disable_comments,
shorten_links=input_data.shorten_links,
unsplash=input_data.unsplash,
requires_approval=input_data.requires_approval,
random_post=input_data.random_post,
random_media_url=input_data.random_media_url,
notes=input_data.notes,
youtube_options=youtube_options,
profile_key=profile_key.get_secret_value(),
)
yield "post_result", response
if response.postIds:
for p in response.postIds:
yield "post", p

3
autogpt_platform/backend/backend/blocks/exa/answers.py
View File

@@ -119,6 +119,3 @@ class ExaAnswerBlock(Block):
except Exception as e:
yield "error", str(e)
yield "answer", ""
yield "citations", []
yield "cost_dollars", {}

1
autogpt_platform/backend/backend/data/block.py
View File

@@ -55,6 +55,7 @@ class BlockType(Enum):
WEBHOOK_MANUAL = "Webhook (manual)"
AGENT = "Agent"
AI = "AI"
AYRSHARE = "Ayrshare"
class BlockCategory(Enum):

19
autogpt_platform/backend/backend/data/model.py
View File

@@ -14,7 +14,6 @@ from typing import (
Generic,
Literal,
Optional,
TypedDict,
TypeVar,
cast,
get_args,
@@ -38,6 +37,7 @@ from pydantic_core import (
ValidationError,
core_schema,
)
from typing_extensions import TypedDict
from backend.integrations.providers import ProviderName
from backend.util.settings import Secrets
@@ -316,15 +316,32 @@ class OAuthState(BaseModel):
class UserMetadata(BaseModel):
integration_credentials: list[Credentials] = Field(default_factory=list)
"""⚠️ Deprecated; use `UserIntegrations.credentials` instead"""
integration_oauth_states: list[OAuthState] = Field(default_factory=list)
"""⚠️ Deprecated; use `UserIntegrations.oauth_states` instead"""
class UserMetadataRaw(TypedDict, total=False):
integration_credentials: list[dict]
"""⚠️ Deprecated; use `UserIntegrations.credentials` instead"""
integration_oauth_states: list[dict]
"""⚠️ Deprecated; use `UserIntegrations.oauth_states` instead"""
class UserIntegrations(BaseModel):
class ManagedCredentials(BaseModel):
"""Integration credentials managed by us, rather than by the user"""
ayrshare_profile_key: Optional[SecretStr] = None
@field_serializer("*")
def dump_secret_strings(value: Any, _info):
if isinstance(value, SecretStr):
return value.get_secret_value()
return value
managed_credentials: ManagedCredentials = Field(default_factory=ManagedCredentials)
credentials: list[Credentials] = Field(default_factory=list)
oauth_states: list[OAuthState] = Field(default_factory=list)

12
autogpt_platform/backend/backend/executor/manager.py
View File

@@ -35,6 +35,7 @@ from autogpt_libs.utils.cache import thread_cached
from prometheus_client import Gauge, start_http_server
from backend.blocks.agent import AgentExecutorBlock
from backend.blocks.ayrshare import AYRSHARE_BLOCK_IDS
from backend.data import redis_client as redis
from backend.data.block import (
BlockData,
@@ -182,6 +183,17 @@ async def execute_node(
)
extra_exec_kwargs[field_name] = credentials
if node_block.id in AYRSHARE_BLOCK_IDS:
profile_key = await creds_manager.store.get_ayrshare_profile_key(user_id)
if not profile_key:
logger.error(
"Ayrshare profile not configured. Please link a social account via Ayrshare integration first."
)
raise ValueError(
"Ayrshare profile not configured. Please link a social account via Ayrshare integration first."
)
extra_exec_kwargs["profile_key"] = profile_key
output_size = 0
try:
async for output_name, output_data in node_block.execute(

515
autogpt_platform/backend/backend/integrations/ayrshare.py Normal file
View File

@@ -0,0 +1,515 @@
from __future__ import annotations
import json
import logging
from enum import Enum
from typing import Any, Optional
from pydantic import BaseModel
from backend.util.exceptions import MissingConfigError
from backend.util.request import Requests
from backend.util.settings import Settings
logger = logging.getLogger(__name__)
settings = Settings()
class AyrshareAPIException(Exception):
def __init__(self, message: str, status_code: int):
super().__init__(message)
self.status_code = status_code
class SocialPlatform(str, Enum):
BLUESKY = "bluesky"
FACEBOOK = "facebook"
TWITTER = "twitter"
LINKEDIN = "linkedin"
INSTAGRAM = "instagram"
YOUTUBE = "youtube"
REDDIT = "reddit"
TELEGRAM = "telegram"
GOOGLE_MY_BUSINESS = "gmb"
PINTEREST = "pinterest"
TIKTOK = "tiktok"
SNAPCHAT = "snapchat"
THREADS = "threads"
class EmailConfig(BaseModel):
to: str
subject: Optional[str] = None
body: Optional[str] = None
from_name: Optional[str] = None
from_email: Optional[str] = None
class JWTResponse(BaseModel):
status: str
title: str
token: str
url: str
emailSent: Optional[bool] = None
expiresIn: Optional[str] = None
class ProfileResponse(BaseModel):
status: str
title: str
refId: str
profileKey: str
messagingActive: Optional[bool] = None
class PostResponse(BaseModel):
status: str
id: str
refId: str
profileTitle: str
post: str
postIds: Optional[list[PostIds]] = None
scheduleDate: Optional[str] = None
errors: Optional[list[str]] = None
class PostIds(BaseModel):
status: str
id: str
postUrl: str
platform: str
class AutoHashtag(BaseModel):
max: Optional[int] = None
position: Optional[str] = None
class FirstComment(BaseModel):
text: str
platforms: Optional[list[SocialPlatform]] = None
class AutoSchedule(BaseModel):
interval: str
platforms: Optional[list[SocialPlatform]] = None
startDate: Optional[str] = None
endDate: Optional[str] = None
class AutoRepost(BaseModel):
interval: str
platforms: Optional[list[SocialPlatform]] = None
startDate: Optional[str] = None
endDate: Optional[str] = None
class AyrshareClient:
"""Client for the Ayrshare Social Media Post API"""
API_URL = "https://api.ayrshare.com/api"
POST_ENDPOINT = f"{API_URL}/post"
PROFILES_ENDPOINT = f"{API_URL}/profiles"
JWT_ENDPOINT = f"{PROFILES_ENDPOINT}/generateJWT"
def __init__(
self,
custom_requests: Optional[Requests] = None,
):
if not settings.secrets.ayrshare_api_key:
raise MissingConfigError("AYRSHARE_API_KEY is not configured")
headers: dict[str, str] = {
"Content-Type": "application/json",
"Authorization": f"Bearer {settings.secrets.ayrshare_api_key}",
}
self.headers = headers
if custom_requests:
self._requests = custom_requests
else:
self._requests = Requests(
extra_headers=headers,
trusted_origins=["https://api.ayrshare.com"],
)
async def generate_jwt(
self,
private_key: str,
profile_key: str,
logout: Optional[bool] = None,
redirect: Optional[str] = None,
allowed_social: Optional[list[SocialPlatform]] = None,
verify: Optional[bool] = None,
base64: Optional[bool] = None,
expires_in: Optional[int] = None,
email: Optional[EmailConfig] = None,
) -> JWTResponse:
"""
Generate a JSON Web Token (JWT) for use with single sign on.
Docs: https://www.ayrshare.com/docs/apis/profiles/generate-jwt-overview
Args:
domain: Domain of app. Must match the domain given during onboarding.
private_key: Private Key used for encryption.
profile_key: User Profile Key (not the API Key).
logout: Automatically logout the current session.
redirect: URL to redirect to when the "Done" button or logo is clicked.
allowed_social: List of social networks to display in the linking page.
verify: Verify that the generated token is valid (recommended for non-production).
base64: Whether the private key is base64 encoded.
expires_in: Token longevity in minutes (1-2880).
email: Configuration for sending Connect Accounts email.
Returns:
JWTResponse object containing the JWT token and URL.
Raises:
AyrshareAPIException: If the API request fails or private key is invalid.
"""
payload: dict[str, Any] = {
"domain": "id-pojeg",
"privateKey": private_key,
"profileKey": profile_key,
}
headers = self.headers
headers["Profile-Key"] = profile_key
if logout is not None:
payload["logout"] = logout
if redirect is not None:
payload["redirect"] = redirect
if allowed_social is not None:
payload["allowedSocial"] = [p.value for p in allowed_social]
if verify is not None:
payload["verify"] = verify
if base64 is not None:
payload["base64"] = base64
if expires_in is not None:
payload["expiresIn"] = expires_in
if email is not None:
payload["email"] = email.model_dump(exclude_none=True)
response = await self._requests.post(
self.JWT_ENDPOINT, json=payload, headers=headers
)
if not response.ok:
try:
error_data = response.json()
error_message = error_data.get("message", "Unknown error")
except json.JSONDecodeError:
error_message = response.text()
raise AyrshareAPIException(
f"Ayrshare API request failed ({response.status}): {error_message}",
response.status,
)
response_data = response.json()
if response_data.get("status") != "success":
raise AyrshareAPIException(
f"Ayrshare API returned error: {response_data.get('message', 'Unknown error')}",
response.status,
)
return JWTResponse(**response_data)
async def create_profile(
self,
title: str,
messaging_active: Optional[bool] = None,
hide_top_header: Optional[bool] = None,
top_header: Optional[str] = None,
disable_social: Optional[list[SocialPlatform]] = None,
team: Optional[bool] = None,
email: Optional[str] = None,
sub_header: Optional[str] = None,
tags: Optional[list[str]] = None,
) -> ProfileResponse:
"""
Create a new User Profile under your Primary Profile.
Docs: https://www.ayrshare.com/docs/apis/profiles/create-profile
Args:
title: Title of the new profile. Must be unique.
messaging_active: Set to true to activate messaging for this user profile.
hide_top_header: Hide the top header on the social accounts linkage page.
top_header: Change the header on the social accounts linkage page.
disable_social: Array of social networks that are disabled for this user's profile.
team: Create a new user profile as a team member.
email: Email address for team member invite (required if team is true).
sub_header: Change the sub header on the social accounts linkage page.
tags: Array of strings to tag user profiles.
Returns:
ProfileResponse object containing the profile details and profile key.
Raises:
AyrshareAPIException: If the API request fails or profile title already exists.
"""
payload: dict[str, Any] = {
"title": title,
}
if messaging_active is not None:
payload["messagingActive"] = messaging_active
if hide_top_header is not None:
payload["hideTopHeader"] = hide_top_header
if top_header is not None:
payload["topHeader"] = top_header
if disable_social is not None:
payload["disableSocial"] = [p.value for p in disable_social]
if team is not None:
payload["team"] = team
if email is not None:
payload["email"] = email
if sub_header is not None:
payload["subHeader"] = sub_header
if tags is not None:
payload["tags"] = tags
response = await self._requests.post(self.PROFILES_ENDPOINT, json=payload)
if not response.ok:
try:
error_data = response.json()
error_message = error_data.get("message", "Unknown error")
except json.JSONDecodeError:
error_message = response.text()
raise AyrshareAPIException(
f"Ayrshare API request failed ({response.status}): {error_message}",
response.status,
)
response_data = response.json()
if response_data.get("status") != "success":
raise AyrshareAPIException(
f"Ayrshare API returned error: {response_data.get('message', 'Unknown error')}",
response.status,
)
return ProfileResponse(**response_data)
async def create_post(
self,
post: str,
platforms: list[SocialPlatform],
*,
media_urls: Optional[list[str]] = None,
is_video: Optional[bool] = None,
schedule_date: Optional[str] = None,
validate_schedule: Optional[bool] = None,
first_comment: Optional[FirstComment] = None,
disable_comments: Optional[bool] = None,
shorten_links: Optional[bool] = None,
auto_schedule: Optional[AutoSchedule] = None,
auto_repost: Optional[AutoRepost] = None,
auto_hashtag: Optional[AutoHashtag | bool] = None,
unsplash: Optional[str] = None,
bluesky_options: Optional[dict[str, Any]] = None,
facebook_options: Optional[dict[str, Any]] = None,
gmb_options: Optional[dict[str, Any]] = None,
instagram_options: Optional[dict[str, Any]] = None,
linkedin_options: Optional[dict[str, Any]] = None,
pinterest_options: Optional[dict[str, Any]] = None,
reddit_options: Optional[dict[str, Any]] = None,
snapchat_options: Optional[dict[str, Any]] = None,
telegram_options: Optional[dict[str, Any]] = None,
threads_options: Optional[dict[str, Any]] = None,
tiktok_options: Optional[dict[str, Any]] = None,
twitter_options: Optional[dict[str, Any]] = None,
youtube_options: Optional[dict[str, Any]] = None,
requires_approval: Optional[bool] = None,
random_post: Optional[bool] = None,
random_media_url: Optional[bool] = None,
idempotency_key: Optional[str] = None,
notes: Optional[str] = None,
profile_key: Optional[str] = None,
) -> PostResponse:
"""
Create a post across multiple social media platforms.
Docs: https://www.ayrshare.com/docs/apis/post/post
Args:
post: The post text to be published - required
platforms: List of platforms to post to (e.g. [SocialPlatform.TWITTER, SocialPlatform.FACEBOOK]) - required
media_urls: Optional list of media URLs to include - required if is_video is true
is_video: Whether the media is a video - default is false (in api docs)
schedule_date: UTC datetime for scheduling (YYYY-MM-DDThh:mm:ssZ) - default is None (in api docs)
validate_schedule: Whether to validate the schedule date - default is false (in api docs)
first_comment: Configuration for first comment - default is None (in api docs)
disable_comments: Whether to disable comments - default is false (in api docs)
shorten_links: Whether to shorten links - default is false (in api docs)
auto_schedule: Configuration for automatic scheduling - default is None (in api docs https://www.ayrshare.com/docs/apis/auto-schedule/overview)
auto_repost: Configuration for automatic reposting - default is None (in api docs https://www.ayrshare.com/docs/apis/post/overview#auto-repost)
auto_hashtag: Configuration for automatic hashtags - default is None (in api docs https://www.ayrshare.com/docs/apis/post/overview#auto-hashtags)
unsplash: Unsplash image configuration - default is None (in api docs https://www.ayrshare.com/docs/apis/post/overview#unsplash)
------------------------------------------------------------
bluesky_options: Bluesky-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/bluesky
facebook_options: Facebook-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/facebook
gmb_options: Google Business Profile options - https://www.ayrshare.com/docs/apis/post/social-networks/google
instagram_options: Instagram-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/instagram
linkedin_options: LinkedIn-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/linkedin
pinterest_options: Pinterest-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/pinterest
reddit_options: Reddit-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/reddit
snapchat_options: Snapchat-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/snapchat
telegram_options: Telegram-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/telegram
threads_options: Threads-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/threads
tiktok_options: TikTok-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/tiktok
twitter_options: Twitter-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/twitter
youtube_options: YouTube-specific options - https://www.ayrshare.com/docs/apis/post/social-networks/youtube
------------------------------------------------------------
requires_approval: Whether to enable approval workflow - default is false (in api docs)
random_post: Whether to generate random post text - default is false (in api docs)
random_media_url: Whether to generate random media - default is false (in api docs)
idempotency_key: Unique ID for the post - default is None (in api docs)
notes: Additional notes for the post - default is None (in api docs)
Returns:
PostResponse object containing the post details and status
Raises:
AyrshareAPIException: If the API request fails
"""
payload: dict[str, Any] = {
"post": post,
"platforms": [p.value for p in platforms],
}
# Add optional parameters if provided
if media_urls:
payload["mediaUrls"] = media_urls
if is_video is not None:
payload["isVideo"] = is_video
if schedule_date:
payload["scheduleDate"] = schedule_date
if validate_schedule is not None:
payload["validateSchedule"] = validate_schedule
if first_comment:
first_comment_dict = first_comment.model_dump(exclude_none=True)
if first_comment.platforms:
first_comment_dict["platforms"] = [
p.value for p in first_comment.platforms
]
payload["firstComment"] = first_comment_dict
if disable_comments is not None:
payload["disableComments"] = disable_comments
if shorten_links is not None:
payload["shortenLinks"] = shorten_links
if auto_schedule:
auto_schedule_dict = auto_schedule.model_dump(exclude_none=True)
if auto_schedule.platforms:
auto_schedule_dict["platforms"] = [
p.value for p in auto_schedule.platforms
]
payload["autoSchedule"] = auto_schedule_dict
if auto_repost:
auto_repost_dict = auto_repost.model_dump(exclude_none=True)
if auto_repost.platforms:
auto_repost_dict["platforms"] = [p.value for p in auto_repost.platforms]
payload["autoRepost"] = auto_repost_dict
if auto_hashtag:
payload["autoHashtag"] = (
auto_hashtag.model_dump(exclude_none=True)
if isinstance(auto_hashtag, AutoHashtag)
else auto_hashtag
)
if unsplash:
payload["unsplash"] = unsplash
if bluesky_options:
payload["blueskyOptions"] = bluesky_options
if facebook_options:
payload["faceBookOptions"] = facebook_options
if gmb_options:
payload["gmbOptions"] = gmb_options
if instagram_options:
payload["instagramOptions"] = instagram_options
if linkedin_options:
payload["linkedInOptions"] = linkedin_options
if pinterest_options:
payload["pinterestOptions"] = pinterest_options
if reddit_options:
payload["redditOptions"] = reddit_options
if snapchat_options:
payload["snapchatOptions"] = snapchat_options
if telegram_options:
payload["telegramOptions"] = telegram_options
if threads_options:
payload["threadsOptions"] = threads_options
if tiktok_options:
payload["tikTokOptions"] = tiktok_options
if twitter_options:
payload["twitterOptions"] = twitter_options
if youtube_options:
payload["youTubeOptions"] = youtube_options
if requires_approval is not None:
payload["requiresApproval"] = requires_approval
if random_post is not None:
payload["randomPost"] = random_post
if random_media_url is not None:
payload["randomMediaUrl"] = random_media_url
if idempotency_key:
payload["idempotencyKey"] = idempotency_key
if notes:
payload["notes"] = notes
headers = self.headers
if profile_key:
headers["Profile-Key"] = profile_key
response = await self._requests.post(
self.POST_ENDPOINT, json=payload, headers=headers
)
logger.warning(f"Ayrshare request: {payload} and headers: {headers}")
if not response.ok:
logger.error(
f"Ayrshare API request failed ({response.status}): {response.text()}"
)
try:
error_data = response.json()
error_message = error_data.get("message", "Unknown error")
except json.JSONDecodeError:
error_message = response.text()
raise AyrshareAPIException(
f"Ayrshare API request failed ({response.status}): {error_message}",
response.status,
)
response_data = response.json()
if response_data.get("status") != "success":
logger.error(
f"Ayrshare API returned error: {response_data.get('message', 'Unknown error')}"
)
raise AyrshareAPIException(
f"Ayrshare API returned error: {response_data.get('message', 'Unknown error')}",
response.status,
)
# Ayrshare returns an array of posts even for single posts
# It seems like there is only ever one post in the array, and within that
# there are multiple postIds
# There is a seperate endpoint for bulk posting, so feels safe to just take
# the first post from the array
if len(response_data["posts"]) == 0:
logger.error("Ayrshare API returned no posts")
raise AyrshareAPIException(
"Ayrshare API returned no posts",
response.status,
)
logger.warn(f"Ayrshare API returned posts: {response_data['posts']}")
return PostResponse(**response_data["posts"][0])

49
autogpt_platform/backend/backend/integrations/credentials_store.py
View File

@@ -1,6 +1,7 @@
import base64
import hashlib
import secrets
from contextlib import asynccontextmanager
from datetime import datetime, timedelta, timezone
from typing import Optional
@@ -240,6 +241,7 @@ class IntegrationCredentialsStore:
return get_service_client(DatabaseManagerAsyncClient)
# =============== USER-MANAGED CREDENTIALS =============== #
async def add_creds(self, user_id: str, credentials: Credentials) -> None:
async with await self.locked_user_integrations(user_id):
if await self.get_creds_by_id(user_id, credentials.id):
@@ -359,6 +361,39 @@ class IntegrationCredentialsStore:
]
await self._set_user_integration_creds(user_id, filtered_credentials)
# ============== SYSTEM-MANAGED CREDENTIALS ============== #
async def get_ayrshare_profile_key(self, user_id: str) -> SecretStr | None:
"""Get the Ayrshare profile key for a user.
The profile key is used to authenticate API requests to Ayrshare's social media posting service.
See https://www.ayrshare.com/docs/apis/profiles/overview for more details.
Args:
user_id: The ID of the user to get the profile key for
Returns:
The profile key as a SecretStr if set, None otherwise
"""
user_integrations = await self._get_user_integrations(user_id)
return user_integrations.managed_credentials.ayrshare_profile_key
async def set_ayrshare_profile_key(self, user_id: str, profile_key: str) -> None:
"""Set the Ayrshare profile key for a user.
The profile key is used to authenticate API requests to Ayrshare's social media posting service.
See https://www.ayrshare.com/docs/apis/profiles/overview for more details.
Args:
user_id: The ID of the user to set the profile key for
profile_key: The profile key to set
"""
_profile_key = SecretStr(profile_key)
async with self.edit_user_integrations(user_id) as user_integrations:
user_integrations.managed_credentials.ayrshare_profile_key = _profile_key
# ===================== OAUTH STATES ===================== #
async def store_state_token(
self, user_id: str, provider: str, scopes: list[str], use_pkce: bool = False
) -> tuple[str, str]:
@@ -375,6 +410,9 @@ class IntegrationCredentialsStore:
scopes=scopes,
)
async with self.edit_user_integrations(user_id) as user_integrations:
user_integrations.oauth_states.append(state)
async with await self.locked_user_integrations(user_id):
user_integrations = await self._get_user_integrations(user_id)
@@ -428,6 +466,17 @@ class IntegrationCredentialsStore:
return None
# =================== GET/SET HELPERS =================== #
@asynccontextmanager
async def edit_user_integrations(self, user_id: str):
async with await self.locked_user_integrations(user_id):
user_integrations = await self._get_user_integrations(user_id)
yield user_integrations # yield to allow edits
await self.db_manager.update_user_integrations(
user_id=user_id, data=user_integrations
)
async def _set_user_integration_creds(
self, user_id: str, credentials: list[Credentials]
) -> None:

97
autogpt_platform/backend/backend/server/integrations/router.py
View File

@@ -1,5 +1,6 @@
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Annotated, Awaitable, List, Literal
from fastapi import (
@@ -12,7 +13,8 @@ from fastapi import (
Request,
status,
)
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, SecretStr
from starlette.status import HTTP_500_INTERNAL_SERVER_ERROR, HTTP_502_BAD_GATEWAY
from backend.data.graph import get_graph, set_node_webhook
from backend.data.integrations import (
@@ -29,6 +31,7 @@ from backend.data.model import (
OAuth2Credentials,
)
from backend.executor.utils import add_graph_execution
from backend.integrations.ayrshare import AyrshareClient, SocialPlatform
from backend.integrations.creds_manager import IntegrationCredentialsManager
from backend.integrations.oauth import CREDENTIALS_BY_PROVIDER, HANDLERS_BY_NAME
from backend.integrations.providers import ProviderName
@@ -39,7 +42,7 @@ from backend.server.integrations.models import (
get_all_provider_names,
)
from backend.server.v2.library.db import set_preset_webhook, update_preset
from backend.util.exceptions import NeedConfirmation, NotFoundError
from backend.util.exceptions import MissingConfigError, NeedConfirmation, NotFoundError
from backend.util.settings import Settings
if TYPE_CHECKING:
@@ -271,6 +274,11 @@ class CredentialsDeletionNeedsConfirmationResponse(BaseModel):
message: str
class AyrshareSSOResponse(BaseModel):
sso_url: str = Field(..., description="The SSO URL for Ayrshare integration")
expires_at: datetime = Field(..., description="ISO timestamp when the URL expires")
@router.delete("/{provider}/credentials/{cred_id}")
async def delete_credentials(
request: Request,
@@ -556,9 +564,90 @@ def _get_provider_oauth_handler(
)
@router.get("/ayrshare/sso_url")
async def get_ayrshare_sso_url(
user_id: Annotated[str, Depends(get_user_id)],
) -> AyrshareSSOResponse:
"""
Generate an SSO URL for Ayrshare social media integration.
Returns:
dict: Contains the SSO URL for Ayrshare integration
"""
try:
client = AyrshareClient()
except MissingConfigError:
raise HTTPException(
status_code=HTTP_500_INTERNAL_SERVER_ERROR,
detail="Ayrshare integration is not configured",
)
# Ayrshare profile key is stored in the credentials store
# It is generated when creating a new profile, if there is no profile key,
# we create a new profile and store the profile key in the credentials store
profile_key = await creds_manager.store.get_ayrshare_profile_key(user_id)
if not profile_key:
logger.debug(f"Creating new Ayrshare profile for user {user_id}")
try:
profile = await client.create_profile(
title=f"User {user_id}", messaging_active=True
)
profile_key = profile.profileKey
await creds_manager.store.set_ayrshare_profile_key(user_id, profile_key)
except Exception as e:
logger.error(f"Error creating Ayrshare profile for user {user_id}: {e}")
raise HTTPException(
status_code=HTTP_502_BAD_GATEWAY,
detail="Failed to create Ayrshare profile",
)
else:
logger.debug(f"Using existing Ayrshare profile for user {user_id}")
profile_key_str = (
profile_key.get_secret_value()
if isinstance(profile_key, SecretStr)
else str(profile_key)
)
private_key = settings.secrets.ayrshare_jwt_key
# Ayrshare JWT expiry is 2880 minutes (48 hours)
max_expiry_minutes = 2880
try:
logger.debug(f"Generating Ayrshare JWT for user {user_id}")
jwt_response = await client.generate_jwt(
private_key=private_key,
profile_key=profile_key_str,
allowed_social=[
# NOTE: We are enabling platforms one at a time
# to speed up the development process
# SocialPlatform.FACEBOOK,
SocialPlatform.TWITTER,
SocialPlatform.LINKEDIN,
# SocialPlatform.INSTAGRAM,
# SocialPlatform.YOUTUBE,
# SocialPlatform.REDDIT,
# SocialPlatform.TELEGRAM,
# SocialPlatform.GOOGLE_MY_BUSINESS,
# SocialPlatform.PINTEREST,
# SocialPlatform.TIKTOK,
# SocialPlatform.BLUESKY,
# SocialPlatform.SNAPCHAT,
# SocialPlatform.THREADS,
],
expires_in=max_expiry_minutes,
verify=True,
)
except Exception as e:
logger.error(f"Error generating Ayrshare JWT for user {user_id}: {e}")
raise HTTPException(
status_code=HTTP_502_BAD_GATEWAY, detail="Failed to generate JWT"
)
expires_at = datetime.now(timezone.utc) + timedelta(minutes=max_expiry_minutes)
return AyrshareSSOResponse(sso_url=jwt_response.url, expires_at=expires_at)
# === PROVIDER DISCOVERY ENDPOINTS ===
@router.get("/providers", response_model=List[str])
async def list_providers() -> List[str]:
"""

3
autogpt_platform/backend/backend/util/settings.py
View File

@@ -495,7 +495,8 @@ class Secrets(UpdateTrackingModel["Secrets"], BaseSettings):
apollo_api_key: str = Field(default="", description="Apollo API Key")
smartlead_api_key: str = Field(default="", description="SmartLead API Key")
zerobounce_api_key: str = Field(default="", description="ZeroBounce API Key")
ayrshare_api_key: str = Field(default="", description="Ayrshare API Key")
ayrshare_jwt_key: str = Field(default="", description="Ayrshare private Key")
# Add more secret fields as needed
model_config = SettingsConfigDict(

184
autogpt_platform/frontend/src/app/api/__generated__/endpoints/files/files.ts generated Normal file
View File

@@ -0,0 +1,184 @@
/**
* Generated by orval v7.10.0 🍺
* Do not edit manually.
* AutoGPT Agent Server
* This server is used to execute agents that are created by the AutoGPT system.
* OpenAPI spec version: 0.1
*/
import { useMutation } from "@tanstack/react-query";
import type {
MutationFunction,
QueryClient,
UseMutationOptions,
UseMutationResult,
} from "@tanstack/react-query";
import type { BodyPostV1UploadFileToCloudStorage } from "../../models/bodyPostV1UploadFileToCloudStorage";
import type { HTTPValidationError } from "../../models/hTTPValidationError";
import type { PostV1UploadFileToCloudStorageParams } from "../../models/postV1UploadFileToCloudStorageParams";
import type { UploadFileResponse } from "../../models/uploadFileResponse";
import { customMutator } from "../../../mutators/custom-mutator";
type SecondParameter<T extends (...args: never) => unknown> = Parameters<T>[1];
/**
* Upload a file to cloud storage and return a storage key that can be used
with FileStoreBlock and AgentFileInputBlock.
Args:
file: The file to upload
user_id: The user ID
provider: Cloud storage provider ("gcs", "s3", "azure")
expiration_hours: Hours until file expires (1-48)
Returns:
Dict containing the cloud storage path and signed URL
* @summary Upload file to cloud storage
*/
export type postV1UploadFileToCloudStorageResponse200 = {
data: UploadFileResponse;
status: 200;
};
export type postV1UploadFileToCloudStorageResponse422 = {
data: HTTPValidationError;
status: 422;
};
export type postV1UploadFileToCloudStorageResponseComposite =
| postV1UploadFileToCloudStorageResponse200
| postV1UploadFileToCloudStorageResponse422;
export type postV1UploadFileToCloudStorageResponse =
postV1UploadFileToCloudStorageResponseComposite & {
headers: Headers;
};
export const getPostV1UploadFileToCloudStorageUrl = (
params?: PostV1UploadFileToCloudStorageParams,
) => {
const normalizedParams = new URLSearchParams();
Object.entries(params || {}).forEach(([key, value]) => {
if (value !== undefined) {
normalizedParams.append(key, value === null ? "null" : value.toString());
}
});
const stringifiedParams = normalizedParams.toString();
return stringifiedParams.length > 0
? `/api/files/upload?${stringifiedParams}`
: `/api/files/upload`;
};
export const postV1UploadFileToCloudStorage = async (
bodyPostV1UploadFileToCloudStorage: BodyPostV1UploadFileToCloudStorage,
params?: PostV1UploadFileToCloudStorageParams,
options?: RequestInit,
): Promise<postV1UploadFileToCloudStorageResponse> => {
const formData = new FormData();
formData.append(`file`, bodyPostV1UploadFileToCloudStorage.file);
return customMutator<postV1UploadFileToCloudStorageResponse>(
getPostV1UploadFileToCloudStorageUrl(params),
{
...options,
method: "POST",
body: formData,
},
);
};
export const getPostV1UploadFileToCloudStorageMutationOptions = <
TError = HTTPValidationError,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>,
TError,
{
data: BodyPostV1UploadFileToCloudStorage;
params?: PostV1UploadFileToCloudStorageParams;
},
TContext
>;
request?: SecondParameter<typeof customMutator>;
}): UseMutationOptions<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>,
TError,
{
data: BodyPostV1UploadFileToCloudStorage;
params?: PostV1UploadFileToCloudStorageParams;
},
TContext
> => {
const mutationKey = ["postV1UploadFileToCloudStorage"];
const { mutation: mutationOptions, request: requestOptions } = options
? options.mutation &&
"mutationKey" in options.mutation &&
options.mutation.mutationKey
? options
: { ...options, mutation: { ...options.mutation, mutationKey } }
: { mutation: { mutationKey }, request: undefined };
const mutationFn: MutationFunction<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>,
{
data: BodyPostV1UploadFileToCloudStorage;
params?: PostV1UploadFileToCloudStorageParams;
}
> = (props) => {
const { data, params } = props ?? {};
return postV1UploadFileToCloudStorage(data, params, requestOptions);
};
return { mutationFn, ...mutationOptions };
};
export type PostV1UploadFileToCloudStorageMutationResult = NonNullable<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>
>;
export type PostV1UploadFileToCloudStorageMutationBody =
BodyPostV1UploadFileToCloudStorage;
export type PostV1UploadFileToCloudStorageMutationError = HTTPValidationError;
/**
* @summary Upload file to cloud storage
*/
export const usePostV1UploadFileToCloudStorage = <
TError = HTTPValidationError,
TContext = unknown,
>(
options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>,
TError,
{
data: BodyPostV1UploadFileToCloudStorage;
params?: PostV1UploadFileToCloudStorageParams;
},
TContext
>;
request?: SecondParameter<typeof customMutator>;
},
queryClient?: QueryClient,
): UseMutationResult<
Awaited<ReturnType<typeof postV1UploadFileToCloudStorage>>,
TError,
{
data: BodyPostV1UploadFileToCloudStorage;
params?: PostV1UploadFileToCloudStorageParams;
},
TContext
> => {
const mutationOptions =
getPostV1UploadFileToCloudStorageMutationOptions(options);
return useMutation(mutationOptions, queryClient);
};

206
autogpt_platform/frontend/src/app/api/__generated__/endpoints/integrations/integrations.ts generated
View File

@@ -21,6 +21,8 @@ import type {
UseQueryResult,
} from "@tanstack/react-query";
import type { AyrshareSSOResponse } from "../../models/ayrshareSSOResponse";
import type { BodyPostV1Callback } from "../../models/bodyPostV1Callback";
import type { CredentialsMetaResponse } from "../../models/credentialsMetaResponse";
@@ -1530,6 +1532,210 @@ export const usePostV1WebhookPing = <
return useMutation(mutationOptions, queryClient);
};
/**
* Generate an SSO URL for Ayrshare social media integration.
Returns:
dict: Contains the SSO URL for Ayrshare integration
* @summary Get Ayrshare Sso Url
*/
export type getV1GetAyrshareSsoUrlResponse200 = {
data: AyrshareSSOResponse;
status: 200;
};
export type getV1GetAyrshareSsoUrlResponseComposite =
getV1GetAyrshareSsoUrlResponse200;
export type getV1GetAyrshareSsoUrlResponse =
getV1GetAyrshareSsoUrlResponseComposite & {
headers: Headers;
};
export const getGetV1GetAyrshareSsoUrlUrl = () => {
return `/api/integrations/ayrshare/sso_url`;
};
export const getV1GetAyrshareSsoUrl = async (
options?: RequestInit,
): Promise<getV1GetAyrshareSsoUrlResponse> => {
return customMutator<getV1GetAyrshareSsoUrlResponse>(
getGetV1GetAyrshareSsoUrlUrl(),
{
...options,
method: "GET",
},
);
};
export const getGetV1GetAyrshareSsoUrlQueryKey = () => {
return [`/api/integrations/ayrshare/sso_url`] as const;
};
export const getGetV1GetAyrshareSsoUrlQueryOptions = <
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(options?: {
query?: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
>;
request?: SecondParameter<typeof customMutator>;
}) => {
const { query: queryOptions, request: requestOptions } = options ?? {};
const queryKey =
queryOptions?.queryKey ?? getGetV1GetAyrshareSsoUrlQueryKey();
const queryFn: QueryFunction<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>
> = ({ signal }) => getV1GetAyrshareSsoUrl({ signal, ...requestOptions });
return { queryKey, queryFn, ...queryOptions } as UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
> & { queryKey: DataTag<QueryKey, TData, TError> };
};
export type GetV1GetAyrshareSsoUrlQueryResult = NonNullable<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>
>;
export type GetV1GetAyrshareSsoUrlQueryError = unknown;
export function useGetV1GetAyrshareSsoUrl<
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(
options: {
query: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
> &
Pick<
DefinedInitialDataOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>
>,
"initialData"
>;
request?: SecondParameter<typeof customMutator>;
},
queryClient?: QueryClient,
): DefinedUseQueryResult<TData, TError> & {
queryKey: DataTag<QueryKey, TData, TError>;
};
export function useGetV1GetAyrshareSsoUrl<
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(
options?: {
query?: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
> &
Pick<
UndefinedInitialDataOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>
>,
"initialData"
>;
request?: SecondParameter<typeof customMutator>;
},
queryClient?: QueryClient,
): UseQueryResult<TData, TError> & {
queryKey: DataTag<QueryKey, TData, TError>;
};
export function useGetV1GetAyrshareSsoUrl<
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(
options?: {
query?: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
>;
request?: SecondParameter<typeof customMutator>;
},
queryClient?: QueryClient,
): UseQueryResult<TData, TError> & {
queryKey: DataTag<QueryKey, TData, TError>;
};
/**
* @summary Get Ayrshare Sso Url
*/
export function useGetV1GetAyrshareSsoUrl<
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(
options?: {
query?: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
>;
request?: SecondParameter<typeof customMutator>;
},
queryClient?: QueryClient,
): UseQueryResult<TData, TError> & {
queryKey: DataTag<QueryKey, TData, TError>;
} {
const queryOptions = getGetV1GetAyrshareSsoUrlQueryOptions(options);
const query = useQuery(queryOptions, queryClient) as UseQueryResult<
TData,
TError
> & { queryKey: DataTag<QueryKey, TData, TError> };
query.queryKey = queryOptions.queryKey;
return query;
}
/**
* @summary Get Ayrshare Sso Url
*/
export const prefetchGetV1GetAyrshareSsoUrlQuery = async <
TData = Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError = unknown,
>(
queryClient: QueryClient,
options?: {
query?: Partial<
UseQueryOptions<
Awaited<ReturnType<typeof getV1GetAyrshareSsoUrl>>,
TError,
TData
>
>;
request?: SecondParameter<typeof customMutator>;
},
): Promise<QueryClient> => {
const queryOptions = getGetV1GetAyrshareSsoUrlQueryOptions(options);
await queryClient.prefetchQuery(queryOptions);
return queryClient;
};
/**
* Get a list of all available provider names.

14
autogpt_platform/frontend/src/app/api/__generated__/models/ayrshareSSOResponse.ts generated Normal file
View File

@@ -0,0 +1,14 @@
/**
* Generated by orval v7.10.0 🍺
* Do not edit manually.
* AutoGPT Agent Server
* This server is used to execute agents that are created by the AutoGPT system.
* OpenAPI spec version: 0.1
*/
export interface AyrshareSSOResponse {
/** The SSO URL for Ayrshare integration */
sso_url: string;
/** ISO timestamp when the URL expires */
expires_at: string;
}

11
autogpt_platform/frontend/src/app/api/__generated__/models/bodyPostV1UploadFileToCloudStorage.ts generated Normal file
View File

@@ -0,0 +1,11 @@
/**
* Generated by orval v7.10.0 🍺
* Do not edit manually.
* AutoGPT Agent Server
* This server is used to execute agents that are created by the AutoGPT system.
* OpenAPI spec version: 0.1
*/
export interface BodyPostV1UploadFileToCloudStorage {
file: Blob;
}

12
autogpt_platform/frontend/src/app/api/__generated__/models/postV1UploadFileToCloudStorageParams.ts generated Normal file
View File

@@ -0,0 +1,12 @@
/**
* Generated by orval v7.10.0 🍺
* Do not edit manually.
* AutoGPT Agent Server
* This server is used to execute agents that are created by the AutoGPT system.
* OpenAPI spec version: 0.1
*/
export type PostV1UploadFileToCloudStorageParams = {
provider?: string;
expiration_hours?: number;
};

15
autogpt_platform/frontend/src/app/api/__generated__/models/uploadFileResponse.ts generated Normal file
View File

@@ -0,0 +1,15 @@
/**
* Generated by orval v7.10.0 🍺
* Do not edit manually.
* AutoGPT Agent Server
* This server is used to execute agents that are created by the AutoGPT system.
* OpenAPI spec version: 0.1
*/
export interface UploadFileResponse {
file_uri: string;
file_name: string;
size: number;
content_type: string;
expires_in_hours: number;
}

120
autogpt_platform/frontend/src/app/api/openapi.json
View File

@@ -443,6 +443,24 @@
}
}
},
"/api/integrations/ayrshare/sso_url": {
"get": {
"tags": ["v1", "integrations"],
"summary": "Get Ayrshare Sso Url",
"description": "Generate an SSO URL for Ayrshare social media integration.\n\nReturns:\n dict: Contains the SSO URL for Ayrshare integration",
"operationId": "getV1GetAyrshareSsoUrl",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": { "$ref": "#/components/schemas/AyrshareSSOResponse" }
}
}
}
}
}
},
"/api/integrations/providers": {
"get": {
"tags": ["v1", "integrations"],
@@ -823,6 +841,64 @@
}
}
},
"/api/files/upload": {
"post": {
"tags": ["v1", "files"],
"summary": "Upload file to cloud storage",
"description": "Upload a file to cloud storage and return a storage key that can be used\nwith FileStoreBlock and AgentFileInputBlock.\n\nArgs:\n file: The file to upload\n user_id: The user ID\n provider: Cloud storage provider (\"gcs\", \"s3\", \"azure\")\n expiration_hours: Hours until file expires (1-48)\n\nReturns:\n Dict containing the cloud storage path and signed URL",
"operationId": "postV1Upload file to cloud storage",
"parameters": [
{
"name": "provider",
"in": "query",
"required": false,
"schema": {
"type": "string",
"default": "gcs",
"title": "Provider"
}
},
{
"name": "expiration_hours",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 24,
"title": "Expiration Hours"
}
}
],
"requestBody": {
"required": true,
"content": {
"multipart/form-data": {
"schema": {
"$ref": "#/components/schemas/Body_postV1Upload_file_to_cloud_storage"
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": { "$ref": "#/components/schemas/UploadFileResponse" }
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": { "$ref": "#/components/schemas/HTTPValidationError" }
}
}
}
}
}
},
"/api/credits": {
"get": {
"tags": ["v1", "credits"],
@@ -3790,6 +3866,24 @@
"required": ["amount", "threshold"],
"title": "AutoTopUpConfig"
},
"AyrshareSSOResponse": {
"properties": {
"sso_url": {
"type": "string",
"title": "Sso Url",
"description": "The SSO URL for Ayrshare integration"
},
"expires_at": {
"type": "string",
"format": "date-time",
"title": "Expires At",
"description": "ISO timestamp when the URL expires"
}
},
"type": "object",
"required": ["sso_url", "expires_at"],
"title": "AyrshareSSOResponse"
},
"BaseGraph-Input": {
"properties": {
"id": { "type": "string", "title": "Id" },
@@ -3934,6 +4028,14 @@
"required": ["type", "data", "data_index"],
"title": "Body_postV1LogRawAnalytics"
},
"Body_postV1Upload_file_to_cloud_storage": {
"properties": {
"file": { "type": "string", "format": "binary", "title": "File" }
},
"type": "object",
"required": ["file"],
"title": "Body_postV1Upload file to cloud storage"
},
"Body_postV2Add_credits_to_user": {
"properties": {
"user_id": { "type": "string", "title": "User Id" },
@@ -6348,6 +6450,24 @@
"required": ["permissions"],
"title": "UpdatePermissionsRequest"
},
"UploadFileResponse": {
"properties": {
"file_uri": { "type": "string", "title": "File Uri" },
"file_name": { "type": "string", "title": "File Name" },
"size": { "type": "integer", "title": "Size" },
"content_type": { "type": "string", "title": "Content Type" },
"expires_in_hours": { "type": "integer", "title": "Expires In Hours" }
},
"type": "object",
"required": [
"file_uri",
"file_name",
"size",
"content_type",
"expires_in_hours"
],
"title": "UploadFileResponse"
},
"UserHistoryResponse": {
"properties": {
"history": {

75
autogpt_platform/frontend/src/components/CustomNode.tsx
View File

@@ -31,7 +31,7 @@ import {
parseKeys,
setNestedProperty,
} from "@/lib/utils";
import { Button } from "@/components/ui/button";
import { Button } from "@/components/atoms/Button/Button";
import { Switch } from "@/components/ui/switch";
import { TextRenderer } from "@/components/ui/render";
import { history } from "./history";
@@ -54,8 +54,10 @@ import {
CopyIcon,
ExitIcon,
} from "@radix-ui/react-icons";
import { Key } from "@phosphor-icons/react";
import useCredits from "@/hooks/useCredits";
import { getV1GetAyrshareSsoUrl } from "@/app/api/__generated__/endpoints/integrations/integrations";
import { toast } from "@/components/molecules/Toast/use-toast";
export type ConnectionData = Array<{
edge_id: string;
@@ -112,6 +114,8 @@ export const CustomNode = React.memo(
const flowContext = useContext(FlowContext);
const api = useBackendAPI();
const { formatCredits } = useCredits();
const [isLoading, setIsLoading] = useState(false);
let nodeFlowId = "";
if (data.uiType === BlockUIType.AGENT) {
@@ -241,6 +245,59 @@ export const CustomNode = React.memo(
return renderHandles(schema.properties);
};
const generateAyrshareSSOHandles = () => {
const handleSSOLogin = async () => {
setIsLoading(true);
try {
const {
data: { sso_url },
} = await getV1GetAyrshareSsoUrl();
const popup = window.open(sso_url, "_blank", "popup=true");
if (!popup) {
throw new Error(
"Please allow popups for this site to be able to login with Ayrshare",
);
}
} catch (error) {
toast({
title: "Error",
description: `Error getting SSO URL: ${error}`,
variant: "destructive",
});
} finally {
setIsLoading(false);
}
};
return (
<div className="flex flex-col gap-2">
<Button
type="button"
variant="outline"
className="w-full"
onClick={handleSSOLogin}
disabled={isLoading}
>
{isLoading ? (
"Loading..."
) : (
<>
<Key className="mr-2 h-4 w-4" />
Connect Social Media Accounts
</>
)}
</Button>
<NodeHandle
title="SSO Token"
keyName="sso_token"
isConnected={false}
schema={{ type: "string" }}
side="right"
/>
</div>
);
};
const generateInputHandles = (
schema: BlockIORootSchema,
nodeType: BlockUIType,
@@ -827,8 +884,18 @@ export const CustomNode = React.memo(
(A Webhook URL will be generated when you save the agent)
</p>
))}
{data.inputSchema &&
generateInputHandles(data.inputSchema, data.uiType)}
{data.uiType === BlockUIType.AYRSHARE ? (
<>
{generateAyrshareSSOHandles()}
{generateInputHandles(
data.inputSchema,
BlockUIType.STANDARD,
)}
</>
) : (
data.inputSchema &&
generateInputHandles(data.inputSchema, data.uiType)
)}
</div>
</div>
) : (

2
autogpt_platform/frontend/src/components/integrations/credentials-provider.tsx
View File

@@ -230,7 +230,7 @@ export default function CredentialsProvider({
useEffect(() => {
if (!isLoggedIn || providerNames.length === 0) {
if (isLoggedIn == false) setProviders(null);
if (isLoggedIn == false) setProviders({});
return;
}

1
autogpt_platform/frontend/src/lib/autogpt-server-api/types.ts
View File

@@ -604,6 +604,7 @@ export enum BlockUIType {
WEBHOOK_MANUAL = "Webhook (manual)",
AGENT = "Agent",
AI = "AI",
AYRSHARE = "Ayrshare",
}
export enum SpecialBlockID {