LinkedIn account-scoped endpoints + DSC/existing-post creatives

briefs/example_linkedin_brief.md

@@ -11,11 +11,11 @@ Generate qualified inbound leads for the Lattice Cloud free-tier signup at
 `lattice.example/cloud`. Hand-off to SDR queue.
 
 ## KPIs
-- **Primary:** Cost per Marketing Qualified Lead (MQL) ≤ $120
+- **Primary:** Cost per Marketing Qualified Lead (MQL) ≤ €110
 - **Secondary:** CTR ≥ 0.6% on Sponsored Content
 
 ## Budget
-$40,000 USD total, across the flight.
+€36,000 EUR total, across the flight.
 
 ## Flight
 2026-07-06 → 2026-08-31 (≈8 weeks).
@@ -57,6 +57,10 @@ Three creatives in market at launch:
 LinkedIn. Mirror flights to Google Search are scoped separately.
 
 ## Notes
+To advertise a post you published by hand on the Company Page, add a line like
+`Existing post: urn:li:share:7340000000000000000` under that creative — the agent
+will reference it directly instead of creating a new Direct Sponsored Content post.
+
 Hold the benchmark-report creative until week 3 so the engineering-leader story
 sets context first. Keep all initial drafts in **DRAFT** status on LinkedIn —
 nothing gets activated in Campaign Manager without sign-off from the demand-gen

docs/linkedin-integration.md

@@ -43,7 +43,7 @@ Optional:
 
 | Variable | Default | Meaning |
 |---|---|---|
-| `LINKEDIN_API_VERSION` | `202405` | Versioned API header (`LinkedIn-Version`). |
+| `LINKEDIN_API_VERSION` | `202605` | Versioned API header (`LinkedIn-Version`). |
 | `LINKEDIN_ALLOWED_AD_ACCOUNTS` | unset | Comma-separated allowlist of ad account ids. The configured account must be in the list, or `YIELDAGENT_ALLOW_LIVE=1` must be set. |
 | `YIELDAGENT_ALLOW_LIVE` | unset | Set to `1` to bypass the allowlist. Disables the only safety net LinkedIn offers through this integration — use sparingly. |
 

src/yieldagent/agents/campaign_setup/prompts.py

@@ -12,6 +12,8 @@
 empty rather than inventing values. Convert currency symbols to ISO codes (e.g. "$" -> "USD").
 Convert dates to ISO 8601. Use lowercase 'meta' for Facebook/Instagram, 'google' for
 Google Ads, 'tiktok' for TikTok.
+If a creative cites the URN of an already-published post (e.g. 'urn:li:share:123' or
+'urn:li:ugcPost:123'), copy it verbatim into that creative's existing_post_urn field.
 """
 
 PLAN_CAMPAIGN_SYSTEM = """\
@@ -28,4 +30,5 @@
 - Carry the Brief's audience through to the LineItem's Targeting unchanged unless the
   Brief specifies sub-audience splits.
 - Use the Brief's notes section to inform naming (e.g. "Midnight Brew Launch — June").
+- Preserve each creative's existing_post_urn unchanged when the Brief sets it.
 """

src/yieldagent/agents/linkedin_setup/cli.py

@@ -22,6 +22,7 @@
 from langgraph.types import Command
 
 from yieldagent.agents.campaign_setup.nodes import DEFAULT_MODEL
+from yieldagent.env import load_dotenv
 
 from .graph import build_graph
 
@@ -119,6 +120,7 @@ async def _run(brief_path: Path, *, auto_approve: bool, dry_run: bool, model_nam
 
 
 def main() -> int:
+    load_dotenv()
     parser = argparse.ArgumentParser(prog="yieldagent-linkedin-setup")
     parser.add_argument("brief", type=Path, help="Path to a markdown campaign brief")
     parser.add_argument(

src/yieldagent/domain/brief.py

@@ -92,6 +92,14 @@ class CreativeAsset(BaseModel):
     video_url: str | None = None
     call_to_action: str | None = None
     landing_url: str | None = None
+    existing_post_urn: str | None = Field(
+        default=None,
+        description=(
+            "URN of an already-published LinkedIn post/share (e.g. 'urn:li:share:123'). "
+            "When set, the ad references this post directly instead of creating a new "
+            "Direct Sponsored Content post — use it to advertise content published by hand."
+        ),
+    )
 
 
 class Brief(BaseModel):

src/yieldagent/integrations/linkedin/client.py

@@ -20,6 +20,9 @@
 
 _BASE_URL = "https://api.linkedin.com/rest"
 _FORBIDDEN_STATUSES = {"ACTIVE", "COMPLETED"}
+# LinkedIn's politicalIntent is a String enum, not a boolean. Sending a bool
+# fails with "enum type is not backed by a String".
+_POLITICAL_INTENT_VALUES = {"POLITICAL", "NOT_POLITICAL", "NOT_DECLARED"}
 
 
 class LinkedInError(RuntimeError):
@@ -37,7 +40,7 @@ def __init__(self, config: LinkedInConfig, http: httpx.AsyncClient | None = None
         self._http = http or httpx.AsyncClient(timeout=30.0)
         self._owns_http = http is None
 
-    async def __aenter__(self) -> "LinkedInClient":
+    async def __aenter__(self) -> LinkedInClient:
         return self
 
     async def __aexit__(self, *_exc: object) -> None:
@@ -147,7 +150,11 @@ async def create_campaign_group(
             payload["totalBudget"] = total_budget
         if run_schedule is not None:
             payload["runSchedule"] = run_schedule
-        return await self._request("POST", "/adCampaignGroups", json=payload)
+        return await self._request(
+            "POST",
+            f"/adAccounts/{self.config.ad_account_id}/adCampaignGroups",
+            json=payload,
+        )
 
     async def create_campaign(
         self,
@@ -164,9 +171,16 @@ async def create_campaign(
         unit_cost: dict[str, str] | None = None,
         cost_type: str = "CPC",
         status: str | None = None,
+        offsite_delivery_enabled: bool = False,
+        political_intent: str = "NOT_POLITICAL",
     ) -> dict[str, Any]:
         if (daily_budget is None) == (total_budget is None):
             raise ValueError("Provide exactly one of daily_budget or total_budget")
+        if political_intent not in _POLITICAL_INTENT_VALUES:
+            raise ValueError(
+                f"political_intent must be one of {sorted(_POLITICAL_INTENT_VALUES)}, "
+                f"got {political_intent!r}"
+            )
         payload: dict[str, Any] = {
             "account": self.config.account_urn,
             "campaignGroup": campaign_group_urn,
@@ -178,26 +192,91 @@ async def create_campaign(
             "runSchedule": run_schedule,
             "targetingCriteria": targeting_criteria,
             "locale": locale,
+            # Both fields became required in current API versions. Safe defaults:
+            # LinkedIn-only delivery (no Audience Network) and non-political.
+            "offsiteDeliveryEnabled": offsite_delivery_enabled,
+            "politicalIntent": political_intent,
         }
         if daily_budget is not None:
             payload["dailyBudget"] = daily_budget
         if total_budget is not None:
             payload["totalBudget"] = total_budget
         if unit_cost is not None:
             payload["unitCost"] = unit_cost
-        return await self._request("POST", "/adCampaigns", json=payload)
+        return await self._request(
+            "POST",
+            f"/adAccounts/{self.config.ad_account_id}/adCampaigns",
+            json=payload,
+        )
+
+    async def create_post(
+        self,
+        *,
+        author_urn: str,
+        commentary: str,
+        article: dict[str, Any] | None = None,
+        dsc_ad_account_urn: str | None = None,
+        feed_distribution: str = "NONE",
+    ) -> dict[str, Any]:
+        """Create a Post via the (non-account-scoped) Posts API.
+
+        A LinkedIn Creative cannot carry inline copy — it must reference a real
+        Post (share / ugcPost). For ads we create a *dark post* (Direct Sponsored
+        Content): authored by the advertiser org, `feedDistribution=NONE` so it
+        never shows on the page's organic feed, and an `adContext` tying it to the
+        sponsored account. Returns the new post URN under `id` (from `x-restli-id`).
+
+        `feedDistribution=NONE` makes LinkedIn treat the post as DSC, which *requires*
+        `adContext.dscAdAccount`. `dscAdType` must NOT be sent — it is read-only and
+        a 422 ("ReadOnly field present in a create request") results otherwise.
+        """
+        payload: dict[str, Any] = {
+            "author": author_urn,
+            "commentary": commentary,
+            "visibility": "PUBLIC",
+            "distribution": {
+                "feedDistribution": feed_distribution,
+                "targetEntities": [],
+                "thirdPartyDistributionChannels": [],
+            },
+            "lifecycleState": "PUBLISHED",
+            "isReshareDisabledByAuthor": False,
+        }
+        if article is not None:
+            payload["content"] = {"article": article}
+        if dsc_ad_account_urn is not None:
+            payload["adContext"] = {"dscAdAccount": dsc_ad_account_urn}
+        return await self._request("POST", "/posts", json=payload)
 
     async def create_creative(
         self,
         *,
         campaign_urn: str,
         content: dict[str, Any],
-        status: str | None = None,
+        intended_status: str | None = None,
     ) -> dict[str, Any]:
+        # The Creatives API rejects `account` (read-only) and uses `intendedStatus`
+        # (an enum) rather than `status`. `content` must reference a Post URN, e.g.
+        # {"reference": "urn:li:share:..."}.
         payload: dict[str, Any] = {
-            "account": self.config.account_urn,
             "campaign": campaign_urn,
             "content": content,
-            "status": self._check_status(status),
+            "intendedStatus": self._check_status(intended_status),
         }
-        return await self._request("POST", "/creatives", json=payload)
+        return await self._request(
+            "POST",
+            f"/adAccounts/{self.config.ad_account_id}/creatives",
+            json=payload,
+        )
+
+    async def list_campaigns(self) -> dict[str, Any]:
+        """List campaigns under the configured ad account.
+
+        Read-only. Useful for smoke tests and for agents that need to know what
+        already exists before planning a new campaign.
+        """
+        return await self._request(
+            "GET",
+            f"/adAccounts/{self.config.ad_account_id}/adCampaigns",
+            params={"q": "search"},
+        )

src/yieldagent/integrations/linkedin/config.py

@@ -13,9 +13,13 @@ class LinkedInConfig:
     api_version: str
     allow_live: bool
     allowed_accounts: frozenset[str] = field(default_factory=frozenset)
+    # Organization (Company Page) that authors the Direct Sponsored Content posts
+    # backing each creative. Optional: if unset, the server resolves it from the
+    # ad account's `reference` field at publish time.
+    organization_urn: str | None = None
 
     @classmethod
-    def from_env(cls) -> "LinkedInConfig":
+    def from_env(cls) -> LinkedInConfig:
         token = os.environ.get("LINKEDIN_ACCESS_TOKEN")
         account = os.environ.get("LINKEDIN_AD_ACCOUNT_ID")
         if not token:
@@ -31,12 +35,19 @@ def from_env(cls) -> "LinkedInConfig":
             )
         raw_allow = os.environ.get("LINKEDIN_ALLOWED_AD_ACCOUNTS", "")
         allowed = frozenset(a.strip() for a in raw_allow.split(",") if a.strip())
+        # Accept a bare org id or a full URN.
+        org = os.environ.get("LINKEDIN_ORGANIZATION_URN") or os.environ.get(
+            "LINKEDIN_ORGANIZATION_ID"
+        )
+        if org and not org.startswith("urn:li:organization:"):
+            org = f"urn:li:organization:{org.strip()}"
         return cls(
             access_token=token,
             ad_account_id=account,
-            api_version=os.environ.get("LINKEDIN_API_VERSION", "202405"),
+            api_version=os.environ.get("LINKEDIN_API_VERSION", "202605"),
             allow_live=os.environ.get("YIELDAGENT_ALLOW_LIVE") == "1",
             allowed_accounts=allowed,
+            organization_urn=org,
         )
 
     @property

src/yieldagent/integrations/linkedin/mapping.py

@@ -73,6 +73,19 @@ def flight_to_run_schedule(flight: Flight) -> dict[str, int]:
     return {"start": int(start.timestamp() * 1000), "end": int(end.timestamp() * 1000)}
 
 
+def campaign_run_schedule(flights: list[Flight]) -> dict[str, int]:
+    """Build a Campaign Group runSchedule that spans all its child Campaigns.
+
+    LinkedIn now requires `runSchedule` on `POST /adAccounts/{id}/adCampaignGroups`.
+    The group must cover the earliest start and latest end across its line items.
+    """
+    if not flights:
+        raise ValueError("Cannot compute runSchedule from empty list of flights")
+    earliest = min(f.start_date for f in flights)
+    latest = max(f.end_date for f in flights)
+    return flight_to_run_schedule(Flight(start_date=earliest, end_date=latest))
+
+
 def audience_to_targeting(audience: Audience) -> dict[str, Any]:
     """Build a LinkedIn `targetingCriteria` payload.
 
@@ -153,12 +166,16 @@ def line_item_payload(
     }
 
 
-def creative_content(creative: CreativeAsset) -> dict[str, Any]:
-    """Build a minimal Sponsored Content `content` block.
+def post_article_content(creative: CreativeAsset) -> dict[str, Any]:
+    """Build the `article` block for a Posts API dark post.
+
+    A Creative can't hold inline copy — it references a Post. We model each ad as
+    an article post pointing at the landing URL. The Posts API does *not* scrape
+    the URL, so we set title/description explicitly.
 
-    Image/video uploads are out of scope for the first slice — the agent will
-    pass `image_url`/`video_url` references; production use will need to upload
-    these assets via the `/assets` endpoint and reference the returned URNs.
+    Thumbnail is intentionally omitted: it must be an `urn:li:image:{id}` from the
+    Images API, not a plain URL. Uploading creative imagery is a follow-up; until
+    then posts render with LinkedIn's default link preview.
     """
     article: dict[str, Any] = {
         "source": creative.landing_url or "https://example.com",
@@ -167,17 +184,17 @@ def creative_content(creative: CreativeAsset) -> dict[str, Any]:
         article["title"] = creative.headline
     if creative.description:
         article["description"] = creative.description
-    if creative.image_url:
-        article["thumbnail"] = creative.image_url
-
-    content: dict[str, Any] = {"article": article}
-    if creative.primary_text:
-        content["commentary"] = creative.primary_text
-    if creative.call_to_action:
-        content["callToAction"] = {
-            "label": creative.call_to_action.upper().replace(" ", "_")
-        }
-    return content
+    return article
+
+
+def post_commentary(creative: CreativeAsset) -> str:
+    """The text shown above the post. Falls back through primary_text → headline → name."""
+    return creative.primary_text or creative.headline or creative.name
+
+
+def creative_content_reference(post_urn: str) -> dict[str, Any]:
+    """Wrap a Post URN as a Creative `content` reference."""
+    return {"reference": post_urn}
 
 
 __all__ = [
@@ -186,9 +203,12 @@ def creative_content(creative: CreativeAsset) -> dict[str, Any]:
     "OBJECTIVE_TO_LINKEDIN",
     "audience_to_targeting",
     "campaign_objective",
-    "creative_content",
+    "campaign_run_schedule",
+    "creative_content_reference",
     "flight_to_run_schedule",
     "line_item_locale",
     "line_item_payload",
     "money_to_linkedin_amount",
+    "post_article_content",
+    "post_commentary",
 ]