Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,14 @@ Through this LinkedIn MCP server, AI assistants like Claude can connect to your

| Tool | Description | Status |
|------|-------------|--------|
| `get_person_profile` | Get profile info with explicit section selection (experience, education, interests, honors, languages, certifications, skills, projects, contact_info, posts) | working |
| `get_person_profile` | Get profile info with explicit section selection (experience, education, interests, honors, languages, certifications, skills, projects, contact_info, posts). The `posts` section surfaces each post's permalink in `references.posts` (kind `feed_post`) so callers can chain to `comment_on_post`. | working |
| `connect_with_person` | Send a connection request or accept an incoming one, with optional note | [#407](https://github.com/stickerdaniel/linkedin-mcp-server/issues/407) |
| `get_sidebar_profiles` | Extract profile URLs from sidebar recommendation sections ("More profiles for you", "Explore premium profiles", "People you may know") on a profile page | working |
| `get_inbox` | List recent conversations from the LinkedIn messaging inbox | working |
| `get_conversation` | Read a specific messaging conversation by username or thread ID | working |
| `search_conversations` | Search messages by keyword | working |
| `send_message` | Send a message to a LinkedIn user (requires confirmation) | working |
| `comment_on_post` | Post a top-level comment on a feed post (requires confirmation) | working |
| `get_company_profile` | Extract company information with explicit section selection (posts, jobs) | working |
| `get_company_posts` | Get recent posts from a company's LinkedIn feed | working |
| `search_jobs` | Search for jobs with keywords and location filters | working |
Expand Down
1 change: 1 addition & 0 deletions docs/docker-hub.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ A Model Context Protocol (MCP) server that connects AI assistants to LinkedIn. A
- **People Search**: Search for people by keywords and location
- **Person Posts**: Get recent activity/posts from a person's profile
- **Company Posts**: Get recent posts from a company's LinkedIn feed
- **Post Engagement**: Post top-level comments on feed posts (requires explicit confirmation)
- **Compact References**: Return typed per-section links alongside readable text without shipping full-page markdown

## Quick Start
Expand Down
419 changes: 418 additions & 1 deletion linkedin_mcp_server/scraping/extractor.py

Large diffs are not rendered by default.

14 changes: 14 additions & 0 deletions linkedin_mcp_server/scraping/link_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,14 @@ class RawReference(TypedDict, total=False):
_NEWSLETTER_PATH_RE = re.compile(r"^/newsletters/([^/?#]+)")
_PULSE_PATH_RE = re.compile(r"^/pulse/([^/?#]+)")
_FEED_PATH_RE = re.compile(r"^/feed/update/([^/?#]+)")
# Share-style post permalinks LinkedIn renders on /recent-activity/all/ pages
# alongside the /feed/update/ form. Shape:
# /posts/<slug>-activity-<19-digit-id>-<short-sig>/
# The 15+ digit floor avoids matching unrelated /posts/<slug>-... URLs that
# happen to contain shorter numeric segments. Both LinkedIn shapes carry the
# same activity id, so we canonicalize to /feed/update/urn:li:activity:{id}/
# and emit kind="feed_post" to fold into the existing reference pipeline.
_SHARE_POST_PATH_RE = re.compile(r"^/posts/[^/?#]*-activity-(\d{15,})-[^/?#]+/?$")
_MESSAGING_THREAD_PATH_RE = re.compile(r"^/messaging/thread/([^/?#]+)")
_MAX_REDIRECT_UNWRAP_DEPTH = 5

Expand Down Expand Up @@ -235,6 +243,12 @@ def classify_link(href: str) -> tuple[ReferenceKind, str] | None:
if match := _FEED_PATH_RE.match(path):
return "feed_post", f"/feed/update/{match.group(1)}/"

if match := _SHARE_POST_PATH_RE.match(path):
return (
"feed_post",
f"/feed/update/urn:li:activity:{match.group(1)}/",
)

if match := _MESSAGING_THREAD_PATH_RE.match(path):
return "conversation", f"/messaging/thread/{match.group(1)}/"

Expand Down
2 changes: 2 additions & 0 deletions linkedin_mcp_server/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
from linkedin_mcp_server.tools.job import register_job_tools
from linkedin_mcp_server.tools.messaging import register_messaging_tools
from linkedin_mcp_server.tools.person import register_person_tools
from linkedin_mcp_server.tools.post import register_post_tools

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -60,6 +61,7 @@ def create_mcp_server(*, tool_timeout: float = DEFAULT_TOOL_TIMEOUT_SECONDS) ->
register_company_tools(mcp, tool_timeout=tool_timeout)
register_job_tools(mcp, tool_timeout=tool_timeout)
register_messaging_tools(mcp, tool_timeout=tool_timeout)
register_post_tools(mcp, tool_timeout=tool_timeout)

# Register session management tool
@mcp.tool(
Expand Down
85 changes: 85 additions & 0 deletions linkedin_mcp_server/tools/post.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
"""LinkedIn feed post engagement tools."""

import logging
from typing import Annotated, Any

from fastmcp import Context, FastMCP
from pydantic import Field

from linkedin_mcp_server.config.schema import DEFAULT_TOOL_TIMEOUT_SECONDS
from linkedin_mcp_server.core.exceptions import AuthenticationError
from linkedin_mcp_server.dependencies import get_ready_extractor, handle_auth_error
from linkedin_mcp_server.error_handler import raise_tool_error

logger = logging.getLogger(__name__)


def register_post_tools(
mcp: FastMCP, *, tool_timeout: float = DEFAULT_TOOL_TIMEOUT_SECONDS
) -> None:
"""Register feed-post engagement tools with the MCP server."""

@mcp.tool(
timeout=tool_timeout,
title="Comment on Post",
annotations={"destructiveHint": True, "openWorldHint": True},
tags={"feed", "actions"},
exclude_args=["extractor"],
)
async def comment_on_post(
post_url: str,
comment_text: Annotated[str, Field(min_length=1, max_length=1500)],
confirm_post: bool,
ctx: Context,
extractor: Any | None = None,
) -> dict[str, Any]:
"""
Post a top-level comment on a LinkedIn feed post.

This is a write operation when confirm_post is True. With
confirm_post=False the composer is located but no comment is
submitted, returning status="confirmation_required" — useful for
dry-run pre-flight checks before issuing the real call.

Args:
post_url: A feed post URL or activity reference. Accepts the
full /feed/update/urn:li:activity:{id}/ URL, the
/posts/<slug>-activity-{id}-<sig>/ permalink, a bare
urn:li:activity:{id} URN, or the bare numeric id.
comment_text: The comment body (1-1500 characters, plain text).
confirm_post: Must be True to actually submit the comment.
ctx: FastMCP context for progress reporting.

Returns:
Dict with url, status, message, posted, comment_visible.
"""
try:
extractor = extractor or await get_ready_extractor(
ctx, tool_name="comment_on_post"
)
logger.info(
"Posting comment on %s (confirm_post=%s, length=%d)",
post_url,
confirm_post,
len(comment_text),
)

await ctx.report_progress(progress=0, total=100, message="Posting comment")

result = await extractor.post_comment(
post_url,
comment_text,
confirm_post=confirm_post,
)

await ctx.report_progress(progress=100, total=100, message="Complete")

return result

except AuthenticationError as e:
try:
await handle_auth_error(e, ctx)
except Exception as relogin_exc:
raise_tool_error(relogin_exc, "comment_on_post")
except Exception as e:
raise_tool_error(e, "comment_on_post") # NoReturn
4 changes: 4 additions & 0 deletions manifest.json
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,10 @@
"name": "send_message",
"description": "Send a message to a LinkedIn user with explicit confirmation and optional profile URN support"
},
{
"name": "comment_on_post",
"description": "Post a top-level comment on a LinkedIn feed post with explicit confirmation"
},
{
"name": "close_session",
"description": "Properly close browser session and clean up resources"
Expand Down
72 changes: 72 additions & 0 deletions tests/test_link_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -578,3 +578,75 @@ def test_inbox_conversation_without_text_still_captured(self):
assert len(references) == 1
assert references[0]["kind"] == "conversation"
assert references[0]["url"] == "/messaging/thread/2-xyz/"


class TestClassifyShareActivityPermalink:
"""Share-style /posts/<slug>-activity-<id>-<sig>/ canonicalize to feed_post."""

_CANONICAL = "/feed/update/urn:li:activity:7000000000000000000/"

def test_share_permalink_classifies_as_feed_post(self):
result = classify_link(
"https://www.linkedin.com/posts/"
"someone-12345_topic-thoughts-activity-7000000000000000000-AbCd/"
)
assert result == ("feed_post", self._CANONICAL)

def test_share_permalink_with_query_string(self):
result = classify_link(
"https://www.linkedin.com/posts/"
"alice-bob_topic-activity-7000000000000000000-XYZ/"
"?utm_source=share"
)
assert result == ("feed_post", self._CANONICAL)

def test_share_permalink_minimal_slug(self):
"""A bare slug (no underscores or hashtags) still matches."""
result = classify_link(
"https://www.linkedin.com/posts/foo-activity-7000000000000000000-XYZ/"
)
assert result == ("feed_post", self._CANONICAL)

def test_short_id_rejected(self):
"""IDs below the 15-digit floor are not treated as activity ids."""
result = classify_link("https://www.linkedin.com/posts/foo-activity-12345-XYZ/")
assert result is None

def test_signature_required(self):
"""The trailing -<sig> segment is required to match."""
# The URL ends after the digits (no trailing -<sig>) — should not
# match the share-permalink shape.
result = classify_link(
"https://www.linkedin.com/posts/foo-activity-7000000000000000000/"
)
assert result is None

def test_unrelated_posts_url_rejected(self):
"""A /posts/<slug>/ without -activity-<id>- doesn't match."""
result = classify_link("https://www.linkedin.com/posts/some-article-headline/")
assert result is None

def test_share_and_canonical_dedupe(self):
"""Anchor with share permalink + anchor with /feed/update/ collapse to one."""
references = build_references(
[
{
"href": (
"https://www.linkedin.com/posts/"
"alice_topic-activity-7000000000000000000-XYZ/"
),
"text": "Alice's post",
},
{
"href": (
"https://www.linkedin.com/feed/update/"
"urn:li:activity:7000000000000000000/"
),
"text": "",
},
],
"posts",
)
assert len(references) == 1
assert references[0]["kind"] == "feed_post"
assert references[0]["url"] == self._CANONICAL
Loading
Loading