Building a LinkedIn Content Agent with MCP Server, OAuth 2.0, and Gemini

1import requests
2from cryptography.fernet import Fernet
3import json
4
5class LinkedInAuth:
6    TOKEN_ENDPOINT = "https://www.linkedin.com/oauth/v2/accessToken"
7    INTROSPECT_ENDPOINT = "https://www.linkedin.com/oauth/v2/introspectToken"
8
9    def __init__(self, client_id: str, client_secret: str, encryption_key: bytes):
10        self.client_id = client_id
11        self.client_secret = client_secret
12        self.fernet = Fernet(encryption_key)
13
14    def get_valid_token(self, stored_token_blob: bytes) -> str:
15        """Decrypt, check expiry, refresh if needed."""
16        token_data = json.loads(self.fernet.decrypt(stored_token_blob))
17
18        # Introspect before use — LinkedIn may invalidate tokens
19        introspect_resp = requests.post(
20            self.INTROSPECT_ENDPOINT,
21            data={
22                "client_id": self.client_id,
23                "client_secret": self.client_secret,
24                "token": token_data["access_token"],
25            }
26        )
27        token_info = introspect_resp.json()
28
29        if not token_info.get("active"):
30            # Refresh using refresh token
31            return self._refresh_token(token_data["refresh_token"])
32
33        return token_data["access_token"]
34
35    def _refresh_token(self, refresh_token: str) -> str:
36        resp = requests.post(self.TOKEN_ENDPOINT, data={
37            "grant_type": "refresh_token",
38            "refresh_token": refresh_token,
39            "client_id": self.client_id,
40            "client_secret": self.client_secret,
41        })
42        new_tokens = resp.json()
43        # Re-encrypt and store
44        self._store_tokens(new_tokens)
45        return new_tokens["access_token"]

1from dataclasses import dataclass
2import math
3from datetime import datetime
4
5@dataclass
6class FeedPost:
7    post_id: str
8    author: str
9    content: str
10    likes: int
11    comments: int
12    shares: int
13    impressions: int
14    published_at: datetime
15    author_connections: int
16
17class FeedScorer:
18    weights = {
19        "engagement_rate": 0.40,  # (likes + comments + shares) / impressions
20        "recency": 0.20,          # exponential decay over 24 hours
21        "author_authority": 0.20, # connection degree + follower count
22        "topic_relevance": 0.20,  # embedding similarity to configured topics
23    }
24
25    def score(self, post: FeedPost, topic_embedding: list[float]) -> float:
26        # Engagement rate (0–1)
27        total_engagement = post.likes + post.comments + post.shares
28        er = total_engagement / max(post.impressions, 1)
29        er_score = min(er * 10, 1.0)  # Normalize: 10% ER = max score
30
31        # Recency: exponential decay, half-life = 12 hours
32        hours_old = (datetime.now() - post.published_at).total_seconds() / 3600
33        recency_score = math.exp(-hours_old / 12)
34
35        # Author authority: log scale on connection count
36        authority_score = min(math.log10(max(post.author_connections, 1)) / 4, 1.0)
37
38        # Topic relevance via embedding cosine similarity
39        post_embedding = self._embed(post.content)
40        topic_score = self._cosine_similarity(post_embedding, topic_embedding)
41
42        return (
43            self.weights["engagement_rate"] * er_score +
44            self.weights["recency"] * recency_score +
45            self.weights["author_authority"] * authority_score +
46            self.weights["topic_relevance"] * topic_score
47        ) * 100

1import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
2import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
3import { z } from 'zod'
4
5const server = new McpServer({ name: 'linkedin-mcp', version: '1.0.0' })
6
7server.tool(
8  'post_to_linkedin',
9  'Create a post on LinkedIn',
10  {
11    content: z.string().max(3000).describe('Post text content'),
12    visibility: z.enum(['PUBLIC', 'CONNECTIONS']).default('PUBLIC'),
13  },
14  async ({ content, visibility }) => {
15    const token = await auth.getValidToken(storedTokenBlob)
16    const response = await fetch('https://api.linkedin.com/v2/ugcPosts', {
17      method: 'POST',
18      headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
19      body: JSON.stringify({
20        author: `urn:li:person:${PERSON_ID}`,
21        lifecycleState: 'PUBLISHED',
22        specificContent: {
23          'com.linkedin.ugc.ShareContent': {
24            shareCommentary: { text: content },
25            shareMediaCategory: 'NONE',
26          },
27        },
28        visibility: { 'com.linkedin.ugc.MemberNetworkVisibility': visibility },
29      }),
30    })
31    const post = await response.json()
32    return { content: [{ type: 'text', text: `Posted: ${post.id}` }] }
33  }
34)
35
36server.tool(
37  'get_post_analytics',
38  'Get engagement metrics for a LinkedIn post',
39  { post_id: z.string() },
40  async ({ post_id }) => {
41    const metrics = await fetchPostMetrics(post_id)
42    return { content: [{ type: 'text', text: JSON.stringify(metrics, null, 2) }] }
43  }
44)

1def post_with_image(token: str, content: str, image_path: str) -> dict:
2    # Step 1: Register upload
3    register_resp = requests.post(
4        "https://api.linkedin.com/v2/assets?action=registerUpload",
5        headers={"Authorization": f"Bearer {token}"},
6        json={
7            "registerUploadRequest": {
8                "recipes": ["urn:li:digitalmediaRecipe:feedshare-image"],
9                "owner": f"urn:li:person:{PERSON_ID}",
10                "serviceRelationships": [{
11                    "relationshipType": "OWNER",
12                    "identifier": "urn:li:userGeneratedContent"
13                }]
14            }
15        }
16    )
17    upload_url = register_resp.json()["value"]["uploadMechanism"][
18        "com.linkedin.digitalmedia.uploading.MediaUploadHttpRequest"
19    ]["uploadUrl"]
20    asset_urn = register_resp.json()["value"]["asset"]
21
22    # Step 2: Upload image binary
23    with open(image_path, "rb") as f:
24        requests.put(upload_url, data=f, headers={"Authorization": f"Bearer {token}"})
25
26    # Step 3: Create post referencing asset URN
27    return create_post(token, content, media_urn=asset_urn)