"""
Async MCP server for LinkedIn post creation with optional OAuth support.
Provides tools for creating, managing, and optimizing LinkedIn posts using
a design system approach with components, themes, and variants.
OAuth Support:
To enable OAuth, set these environment variables:
- LINKEDIN_CLIENT_ID: LinkedIn app client ID
- LINKEDIN_CLIENT_SECRET: LinkedIn app client secret
- LINKEDIN_REDIRECT_URI: OAuth callback URL (default: http://localhost:8000/oauth/callback)
- OAUTH_SERVER_URL: OAuth server base URL (default: http://localhost:8000)
- OAUTH_ENABLED: Enable OAuth (default: true if credentials present)
Note: Uses generic OAuth implementation from chuk-mcp-server.
"""
import os
from typing import Any, Optional
from chuk_mcp_server import ChukMCPServer
from .api import LinkedInClient
from .manager import LinkedInManager
from .manager_factory import ManagerFactory, set_factory
from .tools.composition_tools import register_composition_tools
from .tools.draft_tools import register_draft_tools
from .tools.publishing_tools import register_publishing_tools
from .tools.registry_tools import register_registry_tools
from .tools.theme_tools import register_theme_tools
# Initialize the MCP server with OAuth provider getter
mcp = ChukMCPServer("chuk-mcp-linkedin")
# Initialize manager factory (creates per-user managers)
# Use artifacts by default, configure storage backend via env vars
manager_factory = ManagerFactory(
use_artifacts=True,
artifact_provider=os.getenv("ARTIFACT_PROVIDER", "memory"),
)
set_factory(manager_factory)
# Legacy: Keep a single manager for backward compatibility (will be deprecated)
manager = LinkedInManager()
linkedin_client = LinkedInClient()
# Set OAuth provider getter in the protocol handler (will be populated after setup_oauth)
mcp.protocol.oauth_provider_getter = lambda: get_oauth_provider()
# Global OAuth provider (will be set if OAuth is enabled)
oauth_provider = None
# Global token store - shared across all OAuth operations
# This ensures tokens stored in one context are visible in another
# TODO: Remove when chuk-sessions ships shared_memory provider
_global_token_store = None
# Register tools with the server (tools will use factory internally)
draft_tools = register_draft_tools(mcp)
composition_tools = register_composition_tools(mcp)
theme_tools = register_theme_tools(mcp)
registry_tools = register_registry_tools(mcp)
publishing_tools = register_publishing_tools(mcp, linkedin_client)
# ============================================================================
# OAuth Integration (Optional)
# ============================================================================
def setup_preview_routes() -> None:
"""Set up preview routes for serving HTML previews."""
from chuk_mcp_server.endpoint_registry import http_endpoint_registry
from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse
async def serve_preview(request: Request) -> HTMLResponse | JSONResponse:
"""Serve HTML preview for a draft using shareable preview token."""
preview_token = request.path_params.get("preview_token")
if not preview_token:
return JSONResponse({"error": "preview_token required"}, status_code=400)
# Search across all active user managers for the draft with this token
try:
from .manager_factory import get_factory
factory = get_factory()
active_users = factory.get_active_users()
# Search through all users' managers
for user_id in active_users:
user_manager = factory.get_manager(user_id)
draft = user_manager.get_draft_by_preview_token(preview_token)
if draft:
# Found the draft, generate/retrieve preview
html_content = await user_manager.read_preview_html_async(draft.draft_id)
if not html_content:
return JSONResponse(
{"error": "Failed to generate preview"}, status_code=500
)
return HTMLResponse(content=html_content)
# Token not found in any user's drafts
return JSONResponse(
{"error": "Preview not found. The draft may have been deleted."}, status_code=404
)
except Exception as e:
return JSONResponse({"error": f"Failed to serve preview: {str(e)}"}, status_code=500)
# Register route using the endpoint registry (called before app is created)
http_endpoint_registry.register_endpoint(
path="/preview/{preview_token}",
handler=serve_preview,
methods=["GET"],
name="preview_draft",
description="Preview a draft post in HTML format using shareable token",
)
# ============================================================================
# HTTP Server Setup (Preview Routes + OAuth)
# ============================================================================
def setup_http_server() -> Optional[Any]:
"""Set up HTTP server features: preview routes and optional OAuth."""
# Always setup preview routes for HTTP mode
setup_preview_routes()
# Setup OAuth if credentials are available
return setup_oauth()
def setup_oauth() -> Optional[Any]:
"""Set up OAuth middleware if credentials are available."""
global oauth_provider, _global_token_store
OAUTH_ENABLED = os.getenv("OAUTH_ENABLED", "true").lower() == "true"
OAUTH_MODE = os.getenv("OAUTH_MODE", "linkedin").lower() # linkedin or keycloak
if not OAUTH_ENABLED:
return None
# Import generic OAuth middleware from chuk-mcp-server
from chuk_mcp_server.oauth import OAuthMiddleware, TokenStore
OAUTH_SERVER_URL = os.getenv("OAUTH_SERVER_URL", "http://localhost:8000")
# Create shared token store (used by both modes)
if _global_token_store is None:
_global_token_store = TokenStore(sandbox_id="chuk-mcp-linkedin")
print("✓ Created shared token store for OAuth")
# ============================================================================
# Keycloak Mode
# ============================================================================
if OAUTH_MODE == "keycloak":
KEYCLOAK_BASE_URL = os.getenv("KEYCLOAK_BASE_URL")
KEYCLOAK_REALM = os.getenv("KEYCLOAK_REALM")
KEYCLOAK_PROVIDER_ALIAS = os.getenv("KEYCLOAK_PROVIDER_ALIAS", "linkedin")
if not KEYCLOAK_BASE_URL or not KEYCLOAK_REALM:
print("⚠ Keycloak mode enabled but configuration missing")
print(" Set KEYCLOAK_BASE_URL and KEYCLOAK_REALM")
return None
# Import Keycloak provider
from .oauth.keycloak_provider import KeycloakOAuthProvider
# Create Keycloak OAuth provider
oauth_provider = KeycloakOAuthProvider(
keycloak_base_url=KEYCLOAK_BASE_URL,
realm_name=KEYCLOAK_REALM,
provider_alias=KEYCLOAK_PROVIDER_ALIAS,
oauth_server_url=OAUTH_SERVER_URL,
token_store=_global_token_store,
)
# Initialize OAuth middleware with Keycloak provider
oauth_middleware = OAuthMiddleware(
mcp_server=mcp,
provider=oauth_provider,
oauth_server_url=OAUTH_SERVER_URL,
callback_path="/oauth/callback", # Dummy path - Keycloak handles actual OAuth flow
scopes_supported=[
"linkedin.posts",
"linkedin.profile",
"linkedin.documents",
],
service_documentation="https://github.com/chrishayuk/chuk-mcp-linkedin",
provider_name="Keycloak (LinkedIn)",
)
# Override the protected resource metadata endpoint for Keycloak mode
# Register AFTER middleware to override the default endpoint
@mcp.endpoint("/.well-known/oauth-protected-resource", methods=["GET"])
async def keycloak_protected_resource_metadata(request: Any) -> Any:
"""OAuth Protected Resource Metadata endpoint - points to Keycloak."""
from starlette.responses import JSONResponse
metadata = oauth_provider.get_protected_resource_metadata() # type: ignore[union-attr]
return JSONResponse(metadata, headers={"Access-Control-Allow-Origin": "*"})
print("✓ OAuth enabled - Keycloak mode")
print(f" MCP Resource Server: {OAUTH_SERVER_URL}")
print(f" Keycloak Authorization Server: {KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}")
print(f" Protected Resource: {OAUTH_SERVER_URL}/.well-known/oauth-protected-resource")
print(f" LinkedIn Provider Alias: {KEYCLOAK_PROVIDER_ALIAS}")
print()
print(" ⚠️ Important Keycloak Configuration:")
print(" 1. Enable 'Store Tokens' in LinkedIn Identity Provider settings")
print(" 2. Add 'broker -> read-token' role to users")
print(" 3. Configure LinkedIn as Identity Provider in Keycloak")
return oauth_middleware
# ============================================================================
# LinkedIn Direct Mode (Default)
# ============================================================================
elif OAUTH_MODE == "linkedin":
LINKEDIN_CLIENT_ID = os.getenv("LINKEDIN_CLIENT_ID")
LINKEDIN_CLIENT_SECRET = os.getenv("LINKEDIN_CLIENT_SECRET")
if not LINKEDIN_CLIENT_ID or not LINKEDIN_CLIENT_SECRET:
print("⚠ OAuth disabled - LinkedIn credentials not configured")
print(" Set LINKEDIN_CLIENT_ID and LINKEDIN_CLIENT_SECRET to enable OAuth")
return None
# Import LinkedIn-specific provider
from .oauth.provider import LinkedInOAuthProvider
# Get OAuth configuration from environment
LINKEDIN_REDIRECT_URI = os.getenv(
"LINKEDIN_REDIRECT_URI", "http://localhost:8000/oauth/callback"
)
# Validate credentials aren't test values
if LINKEDIN_CLIENT_ID.startswith("test_") or LINKEDIN_CLIENT_SECRET.startswith("test_"):
print("⚠️ WARNING: Using test LinkedIn credentials!")
print(" OAuth flow will not work with test credentials.")
print(" To use OAuth, obtain real credentials from:")
print(" https://www.linkedin.com/developers/apps")
# Create LinkedIn OAuth provider with SHARED token store
oauth_provider = LinkedInOAuthProvider(
linkedin_client_id=LINKEDIN_CLIENT_ID,
linkedin_client_secret=LINKEDIN_CLIENT_SECRET,
linkedin_redirect_uri=LINKEDIN_REDIRECT_URI,
oauth_server_url=OAUTH_SERVER_URL,
token_store=_global_token_store,
)
# Initialize generic OAuth middleware with LinkedIn provider
oauth_middleware = OAuthMiddleware(
mcp_server=mcp,
provider=oauth_provider,
oauth_server_url=OAUTH_SERVER_URL,
callback_path="/oauth/callback",
scopes_supported=[
"linkedin.posts",
"linkedin.profile",
"linkedin.documents",
],
service_documentation="https://github.com/chrishayuk/chuk-mcp-linkedin",
provider_name="LinkedIn",
)
print("✓ OAuth enabled - LinkedIn direct mode")
print(f" OAuth server: {OAUTH_SERVER_URL}")
print(f" Discovery: {OAUTH_SERVER_URL}/.well-known/oauth-authorization-server")
print(f" Protected Resource: {OAUTH_SERVER_URL}/.well-known/oauth-protected-resource")
return oauth_middleware
else:
print(f"⚠ Unknown OAUTH_MODE: {OAUTH_MODE}")
print(" Valid modes: linkedin, keycloak")
return None
def get_oauth_provider() -> Optional[Any]:
"""Get the global OAuth provider instance."""
return oauth_provider
def get_token_store() -> Optional[Any]:
"""Get the global token store instance."""
return _global_token_store
# Make tools available at module level for easier imports
__all__ = [
"mcp",
"manager",
"linkedin_client",
"draft_tools",
"composition_tools",
"theme_tools",
"registry_tools",
"publishing_tools",
]
