Coding-With-Raptor/Likendin-Agent

1from raw_function import register_image_upload, upload_image, get_user_id2from dotenv import load_dotenv, find_dotenv3from mcp.server.fastmcp import FastMCP4import requests5import os6 7# Create server8mcp = FastMCP(9    name="Linkedin MCP Server",10    description="A basic MCP server for HTTP client demonstration.",11    stateless_http=True,  # Makes the server treat each HTTP request independently12    json_response=True,  # Allows single JSON responses for non-streaming calls (optimization)13    port=8000,14    host="0.0.0.0",15)16 17# Constants18LINKEDIN_ACCESS_TOKEN = os.getenv("LINKEDIN_ACCESS_TOKEN")19 20load_dotenv(find_dotenv())21 22 23HEADERS = {24    "Authorization": f"Bearer {LINKEDIN_ACCESS_TOKEN}",25    "Content-Type": "application/json",26}27 28 29@mcp.tool(30    name="get_user_info",31    description="Retrieves the authenticated user's LinkedIn profile information including their ID, name, and other profile details.",32)33def get_user_info():34    """35    Retrieve LinkedIn user profile information using the /userinfo endpoint.36 37    Returns:38        dict: User profile information or None if an error occurs.39    """40    url = "https://api.linkedin.com/v2/userinfo"41 42    try:43        response = requests.get(url, headers=HEADERS)44        response.raise_for_status()45        return response.json()46    except requests.exceptions.RequestException as e:47        print(f"Error fetching user info: {e}")48        return None49 50 51@mcp.tool(52    name="post_content_linkedin",53    description="Creates a text-only post on LinkedIn with the specified content. The post will be visible to the public.",54)55def post_content_linkedin(content: str):56    """57    Share content on LinkedIn.58 59    Args:60        content (str): The text content to post on LinkedIn.61 62    Returns:63        dict: API response or None if an error occurs.64    """65    user_info = get_user_id()66    if not user_info:67        print("Failed to retrieve user information.")68        return None69 70    author_urn = f"urn:li:person:{user_info['sub']}"71    url = "https://api.linkedin.com/v2/ugcPosts"72    payload = {73        "author": author_urn,74        "lifecycleState": "PUBLISHED",75        "specificContent": {76            "com.linkedin.ugc.ShareContent": {77                "shareCommentary": {"text": content},78                "shareMediaCategory": "NONE",79            }80        },81        "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"},82    }83 84    try:85        response = requests.post(url, headers=HEADERS, json=payload)86        response.raise_for_status()87        return response.json()88    except requests.exceptions.RequestException as e:89        print(f"Error posting content: {e}")90        return None91 92 93@mcp.tool(94    name="create_post_links_image",95    description="Creates a LinkedIn post with both text content and an image. The image can be provided as a local file path or URL, and the post will be visible to the public.",96)97def create_post_links_image(image_path: str, post_text: str):98    """99    Create a post on LinkedIn with the uploaded image.100 101    Args:102        image_path (str): Path to the image file or URL of the image to post.103        post_text (str): The text content to accompany the image in the post.104 105    Returns:106        None: Prints success or error message.107    """108    # Step 1: Register image upload109    upload_url, image_urn = register_image_upload()110    if not (upload_url and image_urn):111        print("Failed to register image upload.")112        return113 114    # Step 2: Upload the image115    if not upload_image(upload_url, image_path):116        print("Failed to upload image.")117        return118 119    # Step 3: Create the post120    user_info = get_user_info()121    if not user_info:122        print("Failed to retrieve user information.")123        return124 125    url = "https://api.linkedin.com/v2/ugcPosts"126    payload = {127        "author": f"urn:li:person:{user_info['sub']}",128        "lifecycleState": "PUBLISHED",129        "specificContent": {130            "com.linkedin.ugc.ShareContent": {131                "shareCommentary": {"text": post_text},132                "shareMediaCategory": "IMAGE",133                "media": [134                    {135                        "status": "READY",136                        "media": image_urn,137                    }138                ],139            }140        },141        "visibility": {"com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"},142    }143    response = requests.post(url, headers=HEADERS, json=payload)144    if response.status_code == 201:145        print("Post created successfully.")146    else:147        print(f"Error creating post: {response.status_code}")148 149 150# 6. Make the server runnable151if __name__ == "__main__":152    print(f"Starting MCP server '{mcp.name}'...")153    # This uses the built-in server runner from the MCP SDK.154    mcp.run(transport="streamable-http")155