mirror of
https://github.com/Significant-Gravitas/AutoGPT.git
synced 2026-04-08 03:00:28 -04:00
feat(blocks): Add WordPress integration with OAuth and create post block (#10464)
This PR adds WordPress integration to AutoGPT platform, enabling users to create posts on WordPress.com and Jetpack-enabled sites. ### Changes 🏗️ **OAuth Implementation:** - Added WordPress OAuth2 handler (`_oauth.py`) supporting both single blog and global access tokens - Implemented OAuth flow without PKCE (as WordPress doesn't require it) - Added token validation endpoint support - Server-side tokens don't expire, eliminating the need for refresh in most cases **API Integration:** - Created WordPress API client (`_api.py`) with Pydantic models for type safety - Implemented `create_post` function with full support for WordPress post features - Added helper functions for token validation and generic API requests - Fixed response models to handle WordPress API's mixed data types **WordPress Block:** - Created `WordPressCreatePostBlock` in `blog.py` with minimal user-facing options - Exposed fields: site, title, content, excerpt, slug, author, categories, tags, featured_image, media_urls - Posts are published immediately by default - Integrated with platform's OAuth credential system ### 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: - [x] OAuth URL generation works correctly for single blog and global access - [x] Token exchange and validation functions handle WordPress API responses - [x] Create post block properly transforms input data to API format - [x] Response models handle mixed data types from WordPress API The WordPress OAuth provider needs to be configured with client ID and secret from WordPress.com application settings.
This commit is contained in:
Swifty
@@ -0,0 +1,3 @@
|
||||
from .blog import WordPressCreatePostBlock
|
||||
|
||||
__all__ = ["WordPressCreatePostBlock"]
|
||||
498
autogpt_platform/backend/backend/blocks/wordpress/_api.py
Normal file
498
autogpt_platform/backend/backend/blocks/wordpress/_api.py
Normal file
@@ -0,0 +1,498 @@
|
||||
from datetime import datetime
|
||||
from enum import Enum
|
||||
from logging import getLogger
|
||||
from typing import Any, Dict, List, Union
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from backend.sdk import BaseModel, Credentials, Requests
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
WORDPRESS_BASE_URL = "https://public-api.wordpress.com/"
|
||||
|
||||
|
||||
class OAuthAuthorizeRequest(BaseModel):
|
||||
"""OAuth authorization request parameters for WordPress.
|
||||
|
||||
Parameters:
|
||||
client_id: Your application's client ID from WordPress.com
|
||||
redirect_uri: The URI for the authorize response redirect. Must exactly match a redirect URI
|
||||
associated with your application.
|
||||
response_type: Can be "code" or "token". "code" should be used for server side applications.
|
||||
scope: A space delimited list of scopes. Optional, defaults to single blog access.
|
||||
blog: Optional blog parameter with the URL or blog ID for a WordPress.com blog or Jetpack site.
|
||||
"""
|
||||
|
||||
client_id: str
|
||||
redirect_uri: str
|
||||
response_type: str = "code"
|
||||
scope: str | None = None
|
||||
blog: str | None = None
|
||||
|
||||
|
||||
class OAuthTokenRequest(BaseModel):
|
||||
"""OAuth token request parameters for WordPress.
|
||||
|
||||
These parameters must be formatted via application/x-www-form-urlencoded encoding.
|
||||
|
||||
Parameters:
|
||||
code: The grant code returned in the redirect. Can only be used once.
|
||||
client_id: Your application's client ID.
|
||||
redirect_uri: The redirect_uri used in the authorization request.
|
||||
client_secret: Your application's client secret.
|
||||
grant_type: The string "authorization_code".
|
||||
"""
|
||||
|
||||
code: str
|
||||
client_id: str
|
||||
redirect_uri: str
|
||||
client_secret: str
|
||||
grant_type: str = "authorization_code"
|
||||
|
||||
|
||||
class OAuthRefreshTokenRequest(BaseModel):
|
||||
"""OAuth token refresh request parameters for WordPress.
|
||||
|
||||
Note: WordPress OAuth2 tokens do not expire when using the "code" response type,
|
||||
so refresh tokens are typically not needed for server-side applications.
|
||||
|
||||
Parameters:
|
||||
refresh_token: The saved refresh token from the previous token grant.
|
||||
client_id: Your application's client ID.
|
||||
client_secret: Your application's client secret.
|
||||
grant_type: The string "refresh_token".
|
||||
"""
|
||||
|
||||
refresh_token: str
|
||||
client_id: str
|
||||
client_secret: str
|
||||
grant_type: str = "refresh_token"
|
||||
|
||||
|
||||
class OAuthTokenResponse(BaseModel):
|
||||
"""OAuth token response from WordPress.
|
||||
|
||||
Successful response has HTTP status code 200 (OK).
|
||||
|
||||
Parameters:
|
||||
access_token: An opaque string. Can be used to make requests to the WordPress API on behalf
|
||||
of the user.
|
||||
blog_id: The ID of the authorized blog.
|
||||
blog_url: The URL of the authorized blog.
|
||||
token_type: The string "bearer".
|
||||
scope: Optional field for global tokens containing the granted scopes.
|
||||
refresh_token: Optional refresh token (typically not provided for server-side apps).
|
||||
expires_in: Optional expiration time (tokens from code flow don't expire).
|
||||
"""
|
||||
|
||||
access_token: str
|
||||
blog_id: str | None = None
|
||||
blog_url: str | None = None
|
||||
token_type: str = "bearer"
|
||||
scope: str | None = None
|
||||
refresh_token: str | None = None
|
||||
expires_in: int | None = None
|
||||
|
||||
|
||||
def make_oauth_authorize_url(
|
||||
client_id: str,
|
||||
redirect_uri: str,
|
||||
scopes: list[str] | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Generate the OAuth authorization URL for WordPress.
|
||||
|
||||
Args:
|
||||
client_id: Your application's client ID from WordPress.com
|
||||
redirect_uri: The URI for the authorize response redirect
|
||||
scopes: Optional list of scopes. Defaults to single blog access if not provided.
|
||||
blog: Optional blog URL or ID for a WordPress.com blog or Jetpack site.
|
||||
|
||||
Returns:
|
||||
The authorization URL that the user should visit
|
||||
"""
|
||||
# Build request parameters
|
||||
params = {
|
||||
"client_id": client_id,
|
||||
"redirect_uri": redirect_uri,
|
||||
"response_type": "code",
|
||||
}
|
||||
|
||||
if scopes:
|
||||
params["scope"] = " ".join(scopes)
|
||||
|
||||
# Build the authorization URL
|
||||
base_url = f"{WORDPRESS_BASE_URL}oauth2/authorize"
|
||||
query_string = urlencode(params)
|
||||
|
||||
return f"{base_url}?{query_string}"
|
||||
|
||||
|
||||
async def oauth_exchange_code_for_tokens(
|
||||
client_id: str,
|
||||
client_secret: str,
|
||||
code: str,
|
||||
redirect_uri: str,
|
||||
) -> OAuthTokenResponse:
|
||||
"""
|
||||
Exchange an authorization code for access token.
|
||||
|
||||
Args:
|
||||
client_id: Your application's client ID.
|
||||
client_secret: Your application's client secret.
|
||||
code: The authorization code returned by WordPress.
|
||||
redirect_uri: The redirect URI used during authorization.
|
||||
|
||||
Returns:
|
||||
Parsed JSON response containing the access token, blog info, etc.
|
||||
"""
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
|
||||
data = OAuthTokenRequest(
|
||||
code=code,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
redirect_uri=redirect_uri,
|
||||
grant_type="authorization_code",
|
||||
).model_dump(exclude_none=True)
|
||||
|
||||
response = await Requests().post(
|
||||
f"{WORDPRESS_BASE_URL}oauth2/token",
|
||||
headers=headers,
|
||||
data=data,
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
return OAuthTokenResponse.model_validate(response.json())
|
||||
raise ValueError(
|
||||
f"Failed to exchange code for tokens: {response.status} {response.text}"
|
||||
)
|
||||
|
||||
|
||||
async def oauth_refresh_tokens(
|
||||
client_id: str,
|
||||
client_secret: str,
|
||||
refresh_token: str,
|
||||
) -> OAuthTokenResponse:
|
||||
"""
|
||||
Refresh an expired access token (for implicit/client-side tokens only).
|
||||
|
||||
Note: Tokens obtained via the "code" flow for server-side applications do not expire.
|
||||
This is primarily used for client-side applications using implicit OAuth.
|
||||
|
||||
Args:
|
||||
client_id: Your application's client ID.
|
||||
client_secret: Your application's client secret.
|
||||
refresh_token: The refresh token previously issued by WordPress.
|
||||
|
||||
Returns:
|
||||
Parsed JSON response containing the new access token.
|
||||
"""
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
|
||||
data = OAuthRefreshTokenRequest(
|
||||
refresh_token=refresh_token,
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
grant_type="refresh_token",
|
||||
).model_dump(exclude_none=True)
|
||||
|
||||
response = await Requests().post(
|
||||
f"{WORDPRESS_BASE_URL}oauth2/token",
|
||||
headers=headers,
|
||||
data=data,
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
return OAuthTokenResponse.model_validate(response.json())
|
||||
raise ValueError(f"Failed to refresh tokens: {response.status} {response.text}")
|
||||
|
||||
|
||||
class TokenInfoResponse(BaseModel):
|
||||
"""Token validation response from WordPress.
|
||||
|
||||
Parameters:
|
||||
client_id: Your application's client ID.
|
||||
user_id: The WordPress.com user ID.
|
||||
blog_id: The blog ID associated with the token.
|
||||
scope: The scope of the token.
|
||||
"""
|
||||
|
||||
client_id: str
|
||||
user_id: str
|
||||
blog_id: str | None = None
|
||||
scope: str | None = None
|
||||
|
||||
|
||||
async def validate_token(
|
||||
client_id: str,
|
||||
token: str,
|
||||
) -> TokenInfoResponse:
|
||||
"""
|
||||
Validate an access token and get associated metadata.
|
||||
|
||||
Args:
|
||||
client_id: Your application's client ID.
|
||||
token: The access token to validate.
|
||||
|
||||
Returns:
|
||||
Token info including user ID, blog ID, and scope.
|
||||
"""
|
||||
|
||||
params = {
|
||||
"client_id": client_id,
|
||||
"token": token,
|
||||
}
|
||||
|
||||
response = await Requests().get(
|
||||
f"{WORDPRESS_BASE_URL}oauth2/token-info",
|
||||
params=params,
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
return TokenInfoResponse.model_validate(response.json())
|
||||
raise ValueError(f"Invalid token: {response.status} {response.text}")
|
||||
|
||||
|
||||
async def make_api_request(
|
||||
endpoint: str,
|
||||
access_token: str,
|
||||
method: str = "GET",
|
||||
data: dict[str, Any] | None = None,
|
||||
params: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Make an authenticated API request to WordPress.
|
||||
|
||||
Args:
|
||||
endpoint: The API endpoint (e.g., "/rest/v1/me/", "/rest/v1/sites/{site_id}/posts/new")
|
||||
access_token: The OAuth access token
|
||||
method: HTTP method (GET, POST, etc.)
|
||||
data: Request body data for POST/PUT requests
|
||||
params: Query parameters
|
||||
|
||||
Returns:
|
||||
JSON response from the API
|
||||
"""
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {access_token}",
|
||||
}
|
||||
|
||||
if data and method in ["POST", "PUT", "PATCH"]:
|
||||
headers["Content-Type"] = "application/json"
|
||||
|
||||
# Ensure endpoint starts with /
|
||||
if not endpoint.startswith("/"):
|
||||
endpoint = f"/{endpoint}"
|
||||
|
||||
url = f"{WORDPRESS_BASE_URL.rstrip('/')}{endpoint}"
|
||||
|
||||
request_method = getattr(Requests(), method.lower())
|
||||
response = await request_method(
|
||||
url,
|
||||
headers=headers,
|
||||
json=data if method in ["POST", "PUT", "PATCH"] else None,
|
||||
params=params,
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
return response.json()
|
||||
raise ValueError(f"API request failed: {response.status} {response.text}")
|
||||
|
||||
|
||||
# Post-related models and functions
|
||||
|
||||
|
||||
class PostStatus(str, Enum):
|
||||
"""WordPress post status options."""
|
||||
|
||||
PUBLISH = "publish"
|
||||
PRIVATE = "private"
|
||||
DRAFT = "draft"
|
||||
PENDING = "pending"
|
||||
FUTURE = "future"
|
||||
AUTO_DRAFT = "auto-draft"
|
||||
|
||||
|
||||
class PostFormat(str, Enum):
|
||||
"""WordPress post format options."""
|
||||
|
||||
STANDARD = "standard"
|
||||
ASIDE = "aside"
|
||||
CHAT = "chat"
|
||||
GALLERY = "gallery"
|
||||
LINK = "link"
|
||||
IMAGE = "image"
|
||||
QUOTE = "quote"
|
||||
STATUS = "status"
|
||||
VIDEO = "video"
|
||||
AUDIO = "audio"
|
||||
|
||||
|
||||
class CreatePostRequest(BaseModel):
|
||||
"""Request model for creating a WordPress post.
|
||||
|
||||
All fields are optional except those you want to set.
|
||||
"""
|
||||
|
||||
# Basic content
|
||||
title: str | None = None
|
||||
content: str | None = None
|
||||
excerpt: str | None = None
|
||||
|
||||
# Post metadata
|
||||
date: datetime | None = None
|
||||
slug: str | None = None
|
||||
author: str | None = None
|
||||
status: PostStatus | None = PostStatus.PUBLISH
|
||||
password: str | None = None
|
||||
sticky: bool | None = False
|
||||
|
||||
# Organization
|
||||
parent: int | None = None
|
||||
type: str | None = "post"
|
||||
categories: List[str] | None = None
|
||||
tags: List[str] | None = None
|
||||
format: PostFormat | None = None
|
||||
|
||||
# Media
|
||||
featured_image: str | None = None
|
||||
media_urls: List[str] | None = None
|
||||
|
||||
# Sharing
|
||||
publicize: bool | None = True
|
||||
publicize_message: str | None = None
|
||||
|
||||
# Engagement
|
||||
likes_enabled: bool | None = None
|
||||
sharing_enabled: bool | None = True
|
||||
discussion: Dict[str, bool] | None = None
|
||||
|
||||
# Page-specific
|
||||
menu_order: int | None = None
|
||||
page_template: str | None = None
|
||||
|
||||
# Advanced
|
||||
metadata: List[Dict[str, Any]] | None = None
|
||||
|
||||
class Config:
|
||||
json_encoders = {datetime: lambda v: v.isoformat()}
|
||||
|
||||
|
||||
class PostAuthor(BaseModel):
|
||||
"""Author information in post response."""
|
||||
|
||||
ID: int
|
||||
login: str
|
||||
email: Union[str, bool, None] = None
|
||||
name: str
|
||||
nice_name: str
|
||||
URL: str | None = None
|
||||
avatar_URL: str | None = None
|
||||
|
||||
|
||||
class PostResponse(BaseModel):
|
||||
"""Response model for a WordPress post."""
|
||||
|
||||
ID: int
|
||||
site_ID: int
|
||||
author: PostAuthor
|
||||
date: datetime
|
||||
modified: datetime
|
||||
title: str
|
||||
URL: str
|
||||
short_URL: str
|
||||
content: str
|
||||
excerpt: str
|
||||
slug: str
|
||||
guid: str
|
||||
status: str
|
||||
sticky: bool
|
||||
password: str | None = ""
|
||||
parent: Union[Dict[str, Any], bool, None] = None
|
||||
type: str
|
||||
discussion: Dict[str, Union[str, bool, int]]
|
||||
likes_enabled: bool
|
||||
sharing_enabled: bool
|
||||
like_count: int
|
||||
i_like: bool
|
||||
is_reblogged: bool
|
||||
is_following: bool
|
||||
global_ID: str
|
||||
featured_image: str | None = None
|
||||
post_thumbnail: Dict[str, Any] | None = None
|
||||
format: str
|
||||
geo: Union[Dict[str, Any], bool, None] = None
|
||||
menu_order: int | None = None
|
||||
page_template: str | None = None
|
||||
publicize_URLs: List[str]
|
||||
terms: Dict[str, Dict[str, Any]]
|
||||
tags: Dict[str, Dict[str, Any]]
|
||||
categories: Dict[str, Dict[str, Any]]
|
||||
attachments: Dict[str, Dict[str, Any]]
|
||||
attachment_count: int
|
||||
metadata: List[Dict[str, Any]]
|
||||
meta: Dict[str, Any]
|
||||
capabilities: Dict[str, bool]
|
||||
revisions: List[int] | None = None
|
||||
other_URLs: Dict[str, Any] | None = None
|
||||
|
||||
|
||||
async def create_post(
|
||||
credentials: Credentials,
|
||||
site: str,
|
||||
post_data: CreatePostRequest,
|
||||
) -> PostResponse:
|
||||
"""
|
||||
Create a new post on a WordPress site.
|
||||
|
||||
Args:
|
||||
site: Site ID or domain (e.g., "myblog.wordpress.com" or "123456789")
|
||||
access_token: OAuth access token
|
||||
post_data: Post data using CreatePostRequest model
|
||||
|
||||
Returns:
|
||||
PostResponse with the created post details
|
||||
"""
|
||||
|
||||
# Convert the post data to a dictionary, excluding None values
|
||||
data = post_data.model_dump(exclude_none=True)
|
||||
|
||||
# Handle special fields that need conversion
|
||||
if "categories" in data and isinstance(data["categories"], list):
|
||||
data["categories"] = ",".join(str(c) for c in data["categories"])
|
||||
|
||||
if "tags" in data and isinstance(data["tags"], list):
|
||||
data["tags"] = ",".join(str(t) for t in data["tags"])
|
||||
|
||||
# Make the API request
|
||||
endpoint = f"/rest/v1.1/sites/{site}/posts/new"
|
||||
|
||||
headers = {
|
||||
"Authorization": credentials.auth_header(),
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
|
||||
response = await Requests().post(
|
||||
f"{WORDPRESS_BASE_URL.rstrip('/')}{endpoint}",
|
||||
headers=headers,
|
||||
data=data,
|
||||
)
|
||||
|
||||
if response.ok:
|
||||
return PostResponse.model_validate(response.json())
|
||||
|
||||
error_data = (
|
||||
response.json()
|
||||
if response.headers.get("content-type", "").startswith("application/json")
|
||||
else {}
|
||||
)
|
||||
error_message = error_data.get("message", response.text)
|
||||
raise ValueError(f"Failed to create post: {response.status} - {error_message}")
|
||||
27
autogpt_platform/backend/backend/blocks/wordpress/_config.py
Normal file
27
autogpt_platform/backend/backend/blocks/wordpress/_config.py
Normal file
@@ -0,0 +1,27 @@
|
||||
import os
|
||||
|
||||
from backend.sdk import BlockCostType, ProviderBuilder
|
||||
|
||||
from ._oauth import WordPressOAuthHandler, WordPressScope
|
||||
|
||||
builder = ProviderBuilder("wordpress").with_base_cost(1, BlockCostType.RUN)
|
||||
|
||||
|
||||
client_id = os.getenv("WORDPRESS_CLIENT_ID")
|
||||
client_secret = os.getenv("WORDPRESS_CLIENT_SECRET")
|
||||
WORDPRESS_OAUTH_IS_CONFIGURED = bool(client_id and client_secret)
|
||||
|
||||
if WORDPRESS_OAUTH_IS_CONFIGURED:
|
||||
builder = builder.with_oauth(
|
||||
WordPressOAuthHandler,
|
||||
scopes=[
|
||||
v.value
|
||||
for v in [
|
||||
WordPressScope.POSTS,
|
||||
]
|
||||
],
|
||||
client_id_env_var="WORDPRESS_CLIENT_ID",
|
||||
client_secret_env_var="WORDPRESS_CLIENT_SECRET",
|
||||
)
|
||||
|
||||
wordpress = builder.build()
|
||||
214
autogpt_platform/backend/backend/blocks/wordpress/_oauth.py
Normal file
214
autogpt_platform/backend/backend/blocks/wordpress/_oauth.py
Normal file
@@ -0,0 +1,214 @@
|
||||
import time
|
||||
from enum import Enum
|
||||
from logging import getLogger
|
||||
from typing import Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
from backend.sdk import BaseOAuthHandler, OAuth2Credentials, ProviderName, SecretStr
|
||||
|
||||
from ._api import (
|
||||
OAuthTokenResponse,
|
||||
TokenInfoResponse,
|
||||
make_oauth_authorize_url,
|
||||
oauth_exchange_code_for_tokens,
|
||||
oauth_refresh_tokens,
|
||||
validate_token,
|
||||
)
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
|
||||
class WordPressScope(str, Enum):
|
||||
"""WordPress OAuth2 scopes.
|
||||
|
||||
Note: If no scope is specified, the token will grant full access to a single blog.
|
||||
Special scopes:
|
||||
- auth: Access to /me endpoints only, primarily for WordPress.com Connect
|
||||
- global: Full access to all blogs the user has on WordPress.com
|
||||
"""
|
||||
|
||||
# Common endpoint-specific scopes
|
||||
POSTS = "posts"
|
||||
COMMENTS = "comments"
|
||||
LIKES = "likes"
|
||||
FOLLOW = "follow"
|
||||
STATS = "stats"
|
||||
USERS = "users"
|
||||
SITES = "sites"
|
||||
MEDIA = "media"
|
||||
|
||||
# Special scopes
|
||||
AUTH = "auth" # Access to /me endpoints only
|
||||
GLOBAL = "global" # Full access to all user's blogs
|
||||
|
||||
|
||||
class WordPressOAuthHandler(BaseOAuthHandler):
|
||||
"""
|
||||
OAuth2 handler for WordPress.com and Jetpack sites.
|
||||
|
||||
Supports both single blog and global access tokens.
|
||||
Server-side tokens (using 'code' response type) do not expire.
|
||||
"""
|
||||
|
||||
PROVIDER_NAME = ProviderName("wordpress")
|
||||
# Default to no scopes for single blog access
|
||||
DEFAULT_SCOPES = []
|
||||
|
||||
def __init__(self, client_id: str, client_secret: Optional[str], redirect_uri: str):
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret
|
||||
self.redirect_uri = redirect_uri
|
||||
self.scopes = self.DEFAULT_SCOPES
|
||||
|
||||
def get_login_url(
|
||||
self, scopes: list[str], state: str, code_challenge: Optional[str] = None
|
||||
) -> str:
|
||||
logger.debug("Generating WordPress OAuth login URL")
|
||||
# WordPress doesn't require PKCE, so code_challenge is not used
|
||||
if not scopes:
|
||||
logger.debug("No scopes provided, will default to single blog access")
|
||||
scopes = self.scopes
|
||||
|
||||
logger.debug(f"Using scopes: {scopes}")
|
||||
logger.debug(f"State: {state}")
|
||||
|
||||
try:
|
||||
base_url = make_oauth_authorize_url(
|
||||
self.client_id, self.redirect_uri, scopes if scopes else None
|
||||
)
|
||||
|
||||
separator = "&" if "?" in base_url else "?"
|
||||
url = f"{base_url}{separator}state={quote(state)}"
|
||||
logger.debug(f"Generated OAuth URL: {url}")
|
||||
return url
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to generate OAuth URL: {str(e)}")
|
||||
raise
|
||||
|
||||
async def exchange_code_for_tokens(
|
||||
self, code: str, scopes: list[str], code_verifier: Optional[str] = None
|
||||
) -> OAuth2Credentials:
|
||||
logger.debug("Exchanging authorization code for tokens")
|
||||
logger.debug(f"Code: {code[:4]}...")
|
||||
logger.debug(f"Scopes: {scopes}")
|
||||
|
||||
# WordPress doesn't use PKCE, so code_verifier is not needed
|
||||
|
||||
try:
|
||||
response: OAuthTokenResponse = await oauth_exchange_code_for_tokens(
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret if self.client_secret else "",
|
||||
code=code,
|
||||
redirect_uri=self.redirect_uri,
|
||||
)
|
||||
logger.info("Successfully exchanged code for tokens")
|
||||
|
||||
# Store blog info in metadata
|
||||
metadata = {}
|
||||
if response.blog_id:
|
||||
metadata["blog_id"] = response.blog_id
|
||||
if response.blog_url:
|
||||
metadata["blog_url"] = response.blog_url
|
||||
|
||||
# WordPress tokens from code flow don't expire
|
||||
credentials = OAuth2Credentials(
|
||||
access_token=SecretStr(response.access_token),
|
||||
refresh_token=(
|
||||
SecretStr(response.refresh_token)
|
||||
if response.refresh_token
|
||||
else None
|
||||
),
|
||||
access_token_expires_at=None,
|
||||
refresh_token_expires_at=None,
|
||||
provider=self.PROVIDER_NAME,
|
||||
scopes=scopes if scopes else [],
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
if response.expires_in:
|
||||
logger.debug(
|
||||
f"Token expires in {response.expires_in} seconds (client-side token)"
|
||||
)
|
||||
else:
|
||||
logger.debug("Token does not expire (server-side token)")
|
||||
|
||||
return credentials
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to exchange code for tokens: {str(e)}")
|
||||
raise
|
||||
|
||||
async def _refresh_tokens(
|
||||
self, credentials: OAuth2Credentials
|
||||
) -> OAuth2Credentials:
|
||||
"""
|
||||
Added for completeness, as WordPress tokens don't expire
|
||||
"""
|
||||
|
||||
logger.debug("Attempting to refresh OAuth tokens")
|
||||
|
||||
# Server-side tokens don't expire
|
||||
if credentials.access_token_expires_at is None:
|
||||
logger.info("Token does not expire (server-side token), no refresh needed")
|
||||
return credentials
|
||||
|
||||
if credentials.refresh_token is None:
|
||||
logger.error("Cannot refresh tokens - no refresh token available")
|
||||
raise ValueError("No refresh token available")
|
||||
|
||||
try:
|
||||
response: OAuthTokenResponse = await oauth_refresh_tokens(
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret if self.client_secret else "",
|
||||
refresh_token=credentials.refresh_token.get_secret_value(),
|
||||
)
|
||||
logger.info("Successfully refreshed tokens")
|
||||
|
||||
# Preserve blog info from original credentials
|
||||
metadata = credentials.metadata or {}
|
||||
if response.blog_id:
|
||||
metadata["blog_id"] = response.blog_id
|
||||
if response.blog_url:
|
||||
metadata["blog_url"] = response.blog_url
|
||||
|
||||
new_credentials = OAuth2Credentials(
|
||||
access_token=SecretStr(response.access_token),
|
||||
refresh_token=(
|
||||
SecretStr(response.refresh_token)
|
||||
if response.refresh_token
|
||||
else credentials.refresh_token
|
||||
),
|
||||
access_token_expires_at=(
|
||||
int(time.time()) + response.expires_in
|
||||
if response.expires_in
|
||||
else None
|
||||
),
|
||||
refresh_token_expires_at=None,
|
||||
provider=self.PROVIDER_NAME,
|
||||
scopes=credentials.scopes,
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
if response.expires_in:
|
||||
logger.debug(
|
||||
f"New access token expires in {response.expires_in} seconds"
|
||||
)
|
||||
else:
|
||||
logger.debug("New token does not expire")
|
||||
|
||||
return new_credentials
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to refresh tokens: {str(e)}")
|
||||
raise
|
||||
|
||||
async def revoke_tokens(self, credentials: OAuth2Credentials) -> bool:
|
||||
logger.debug("Token revocation requested")
|
||||
logger.info(
|
||||
"WordPress doesn't provide a token revocation endpoint - server-side tokens don't expire"
|
||||
)
|
||||
return False
|
||||
|
||||
async def validate_access_token(self, token: str) -> TokenInfoResponse:
|
||||
"""Validate an access token and get associated metadata."""
|
||||
return await validate_token(self.client_id, token)
|
||||
92
autogpt_platform/backend/backend/blocks/wordpress/blog.py
Normal file
92
autogpt_platform/backend/backend/blocks/wordpress/blog.py
Normal file
@@ -0,0 +1,92 @@
|
||||
from backend.sdk import (
|
||||
Block,
|
||||
BlockCategory,
|
||||
BlockOutput,
|
||||
BlockSchema,
|
||||
Credentials,
|
||||
CredentialsMetaInput,
|
||||
SchemaField,
|
||||
)
|
||||
|
||||
from ._api import CreatePostRequest, PostResponse, PostStatus, create_post
|
||||
from ._config import wordpress
|
||||
|
||||
|
||||
class WordPressCreatePostBlock(Block):
|
||||
"""
|
||||
Creates a new post on a WordPress.com site or Jetpack-enabled site and publishes it.
|
||||
"""
|
||||
|
||||
class Input(BlockSchema):
|
||||
credentials: CredentialsMetaInput = wordpress.credentials_field()
|
||||
site: str = SchemaField(
|
||||
description="Site ID or domain (e.g., 'myblog.wordpress.com' or '123456789')"
|
||||
)
|
||||
title: str = SchemaField(description="The post title")
|
||||
content: str = SchemaField(description="The post content (HTML supported)")
|
||||
excerpt: str | None = SchemaField(
|
||||
description="An optional post excerpt/summary", default=None
|
||||
)
|
||||
slug: str | None = SchemaField(
|
||||
description="The URL slug for the post (auto-generated if not provided)",
|
||||
default=None,
|
||||
)
|
||||
author: str | None = SchemaField(
|
||||
description="Username or ID of the author (defaults to authenticated user)",
|
||||
default=None,
|
||||
)
|
||||
categories: list[str] = SchemaField(
|
||||
description="List of category names or IDs", default=[]
|
||||
)
|
||||
tags: list[str] = SchemaField(
|
||||
description="List of tag names or IDs", default=[]
|
||||
)
|
||||
featured_image: str | None = SchemaField(
|
||||
description="Post ID of an existing attachment to set as featured image",
|
||||
default=None,
|
||||
)
|
||||
media_urls: list[str] = SchemaField(
|
||||
description="URLs of images to sideload and attach to the post", default=[]
|
||||
)
|
||||
|
||||
class Output(BlockSchema):
|
||||
post_id: int = SchemaField(description="The ID of the created post")
|
||||
post_url: str = SchemaField(description="The full URL of the created post")
|
||||
short_url: str = SchemaField(description="The shortened wp.me URL")
|
||||
post_data: dict = SchemaField(description="Complete post data returned by API")
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
id="ee4fe08c-18f9-442f-a985-235379b932e1",
|
||||
description="Create a new post on WordPress.com or Jetpack sites",
|
||||
categories={BlockCategory.SOCIAL},
|
||||
input_schema=self.Input,
|
||||
output_schema=self.Output,
|
||||
)
|
||||
|
||||
async def run(
|
||||
self, input_data: Input, *, credentials: Credentials, **kwargs
|
||||
) -> BlockOutput:
|
||||
post_request = CreatePostRequest(
|
||||
title=input_data.title,
|
||||
content=input_data.content,
|
||||
excerpt=input_data.excerpt,
|
||||
slug=input_data.slug,
|
||||
author=input_data.author,
|
||||
categories=input_data.categories,
|
||||
tags=input_data.tags,
|
||||
featured_image=input_data.featured_image,
|
||||
media_urls=input_data.media_urls,
|
||||
status=PostStatus.PUBLISH,
|
||||
)
|
||||
|
||||
post_response: PostResponse = await create_post(
|
||||
credentials=credentials,
|
||||
site=input_data.site,
|
||||
post_data=post_request,
|
||||
)
|
||||
|
||||
yield "post_id", post_response.ID
|
||||
yield "post_url", post_response.URL
|
||||
yield "short_url", post_response.short_URL
|
||||
yield "post_data", post_response.model_dump()
|
||||
Reference in New Issue
Block a user