diff --git a/briefs/linkedin_live_test_brief.md b/briefs/linkedin_live_test_brief.md new file mode 100644 index 0000000..1cadd4b --- /dev/null +++ b/briefs/linkedin_live_test_brief.md @@ -0,0 +1,43 @@ +# Campaign Brief — TensorOps AdTech Demo (LIVE TEST) + +## Advertiser +TensorOps + +## Product +AI agents for adtech / media buying. + +## Objective +Drive awareness and engagement for the June 17 TensorOps adtech session. + +## KPIs +- **Primary:** Cost per engagement +- **Secondary:** CTR ≥ 0.5% on Sponsored Content + +## Budget +€500 EUR total, across the flight. + +## Flight +2026-06-16 → 2026-06-30 (about two weeks). + +## Platforms +LinkedIn only. + +## Audience +Marketing and advertising decision-makers in the US, UK, and Portugal who work +in adtech, media buying, or programmatic advertising. + +## Creatives +One creative at launch: + +### Creative 1 — June 17 session promo +- **Headline:** How are digital ads chosen in milliseconds? +- **Primary text:** Join TensorOps on June 17 to see how agents run programmatic media buying. +- **Call to action:** Learn More +- **Landing URL:** tensorops.ai +- **Existing post:** urn:li:share:7459747919521333248 + +## Notes +This creative advertises a post already published by hand on the TensorOps +Company Page (referenced above), so the agent references it directly instead of +minting a new Direct Sponsored Content post. Keep everything in **DRAFT** — +nothing gets activated in Campaign Manager without sign-off. diff --git a/src/yieldagent/integrations/linkedin/client.py b/src/yieldagent/integrations/linkedin/client.py index 4cc7557..9da29ba 100644 --- a/src/yieldagent/integrations/linkedin/client.py +++ b/src/yieldagent/integrations/linkedin/client.py @@ -297,6 +297,31 @@ async def delete_post(self, post_urn: str) -> None: key = quote(str(post_urn), safe="") await self._request("DELETE", f"/posts/{key}") + async def typeahead_targeting_entities(self, *, facet: str, query: str) -> list[dict[str, Any]]: + """Search a targeting facet's open taxonomy (industries, titles, skills). + + Returns the relevance-ranked entities `[{urn, name, facetUrn}, ...]`. Only + facets whose `availableEntityFinders` include `TYPEAHEAD` accept this — the + closed enums (seniorities, jobFunctions) 400 here and must use the + standardized-data endpoints instead. + """ + res = await self._request( + "GET", + "/adTargetingEntities", + params={"q": "typeahead", "query": query, "facet": facet}, + ) + return res.get("elements", []) + + async def list_seniorities(self) -> list[dict[str, Any]]: + """Standardized seniority taxonomy: `[{id, name:{localized:{en_US}}}, ...]` (10).""" + res = await self._request("GET", "/seniorities", params={"count": 50}) + return res.get("elements", []) + + async def list_functions(self) -> list[dict[str, Any]]: + """Standardized job-function taxonomy: `[{id, name:{localized:{en_US}}}, ...]` (26).""" + res = await self._request("GET", "/functions", params={"count": 50}) + return res.get("elements", []) + async def list_campaigns(self) -> dict[str, Any]: """List campaigns under the configured ad account. diff --git a/src/yieldagent/integrations/linkedin/mapping.py b/src/yieldagent/integrations/linkedin/mapping.py index 55ab825..c51b9c1 100644 --- a/src/yieldagent/integrations/linkedin/mapping.py +++ b/src/yieldagent/integrations/linkedin/mapping.py @@ -87,14 +87,12 @@ def campaign_run_schedule(flights: list[Flight]) -> dict[str, int]: def audience_to_targeting(audience: Audience) -> dict[str, Any]: - """Build a LinkedIn `targetingCriteria` payload. - - Only geo targeting is wired in this slice. B2B facets (industries, job - functions, seniorities, company sizes, skills) require URN resolution via - the typeahead endpoint — a follow-up. They are present on the Audience for - Brief round-trip fidelity, but are not pushed to the API here. If any B2B - facets are set, a hint is included in the payload under `_unresolved_b2b` - so the caller can log and surface them at approval time. + """Build a geo-only LinkedIn `targetingCriteria` payload. + + This is the static, client-free baseline (locations only). Live B2B facet + resolution — industries, job functions, titles, seniorities, company sizes, + skills — needs API lookups and lives in `targeting.TargetingResolver`, which + the publish flow uses. This helper remains for inspection/test scaffolding. """ includes: list[str] = [] for code in audience.geos: @@ -105,7 +103,7 @@ def audience_to_targeting(audience: Audience) -> dict[str, Any]: # LinkedIn requires at least one location; default to US. includes.append(ISO_TO_LINKEDIN_GEO_URN["US"]) - criteria: dict[str, Any] = { + return { "include": { "and": [ { @@ -117,22 +115,6 @@ def audience_to_targeting(audience: Audience) -> dict[str, Any]: } } - unresolved = { - k: v - for k, v in { - "industries": audience.industries, - "job_functions": audience.job_functions, - "job_titles": audience.job_titles, - "seniorities": audience.seniorities, - "company_sizes": audience.company_sizes, - "skills": audience.skills, - }.items() - if v - } - if unresolved: - criteria["_unresolved_b2b"] = unresolved - return criteria - def line_item_locale(audience: Audience) -> dict[str, str]: """LinkedIn campaigns require a `locale` (country + language). diff --git a/src/yieldagent/integrations/linkedin/server.py b/src/yieldagent/integrations/linkedin/server.py index 83babed..fc37aa8 100644 --- a/src/yieldagent/integrations/linkedin/server.py +++ b/src/yieldagent/integrations/linkedin/server.py @@ -22,7 +22,6 @@ from .config import LinkedInConfig from .mapping import ( DEFAULT_CAMPAIGN_TYPE, - audience_to_targeting, campaign_objective, campaign_run_schedule, creative_content_reference, @@ -32,6 +31,7 @@ post_article_content, post_commentary, ) +from .targeting import TargetingResolver mcp = FastMCP("yieldagent-linkedin") @@ -40,13 +40,6 @@ def _client() -> LinkedInClient: return LinkedInClient(LinkedInConfig.from_env()) -def _strip_unresolved(targeting: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: - """Split the mapping's targeting payload into wire payload + unresolved B2B notes.""" - wire = {k: v for k, v in targeting.items() if not k.startswith("_")} - unresolved = targeting.get("_unresolved_b2b", {}) - return wire, unresolved - - class _Created: """Tracks resources created during a publish so they can be rolled back. @@ -218,12 +211,11 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: line_item_urns: dict[str, str] = {} unresolved_by_li: dict[str, dict[str, Any]] = {} + resolver = TargetingResolver(client) for li in parsed.line_items: - targeting, unresolved = _strip_unresolved( - audience_to_targeting(li.targeting.audience) - ) - if unresolved: - unresolved_by_li[li.name] = unresolved + resolved = await resolver.resolve(li.targeting.audience) + if resolved.unresolved: + unresolved_by_li[li.name] = resolved.unresolved run_schedule = flight_to_run_schedule(li.flight) created_li = await client.create_campaign( campaign_group_urn=group_urn, @@ -232,7 +224,7 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: campaign_type=DEFAULT_CAMPAIGN_TYPE, total_budget=money_to_linkedin_amount(li.budget.amount, li.budget.currency), run_schedule=run_schedule, - targeting_criteria=targeting, + targeting_criteria=resolved.criteria, locale=line_item_locale(li.targeting.audience), ) created.campaigns.append(created_li["id"]) @@ -302,9 +294,10 @@ async def publish_draft_campaign(campaign: dict[str, Any]) -> dict[str, Any]: if unresolved_by_li: result["notes"]["unresolved_b2b_targeting"] = unresolved_by_li result["notes"]["unresolved_b2b_hint"] = ( - "These B2B facets are present on the Brief audience but were not pushed to " - "LinkedIn — they require URN resolution via the typeahead endpoint, which is " - "not wired in this slice. Add them manually in Campaign Manager before activation." + "These facet values came from the Brief but matched no LinkedIn targeting " + "entity (typeahead/standardized lookup returned nothing), so they were not " + "pushed — we never guess a URN. Add them manually in Campaign Manager before " + "activation, or refine the wording to match LinkedIn's taxonomy." ) return result diff --git a/src/yieldagent/integrations/linkedin/targeting.py b/src/yieldagent/integrations/linkedin/targeting.py new file mode 100644 index 0000000..1519100 --- /dev/null +++ b/src/yieldagent/integrations/linkedin/targeting.py @@ -0,0 +1,230 @@ +"""Resolve Audience B2B facets into LinkedIn `targetingCriteria` URNs. + +An LLM extracts free-text facet names from a brief ("Marketing", "Director", +"Programmatic Advertising"); LinkedIn's `targetingCriteria` needs entity URNs. +This module bridges the two, choosing a resolution path per facet based on what +the API exposes (the `availableEntityFinders` on `GET /adTargetingFacets`): + + * Closed enums — seniorities, jobFunctions — do NOT support the typeahead + finder. Their full taxonomy (with display names) comes from the + standardized-data endpoints `/seniorities` and `/functions`; we fetch once + and match names against it. URNs are `urn:li:seniority:{id}` / `urn:li:function:{id}`. + * Open taxonomies — industries, titles, skills — are resolved via the + typeahead finder, which returns `{urn, name}` ranked by relevance. + * company_sizes — a fixed 9-bucket enum whose targeting values are plain + strings (`SIZE_11_TO_50`, ...), mapped statically. + * locations — geo URNs from the static ISO map in `mapping`. + +Names that resolve to no real LinkedIn URN are never guessed: they are collected +in `ResolvedTargeting.unresolved` so the caller can surface them as a manual step +in Campaign Manager. Every URN that *is* used was returned by LinkedIn itself. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Protocol + +from yieldagent.domain import Audience + +from .mapping import ISO_TO_LINKEDIN_GEO_URN + +FACET_LOCATIONS = "urn:li:adTargetingFacet:locations" +FACET_SENIORITIES = "urn:li:adTargetingFacet:seniorities" +FACET_JOB_FUNCTIONS = "urn:li:adTargetingFacet:jobFunctions" +FACET_INDUSTRIES = "urn:li:adTargetingFacet:industries" +FACET_TITLES = "urn:li:adTargetingFacet:titles" +FACET_SKILLS = "urn:li:adTargetingFacet:skills" +FACET_STAFF_COUNT = "urn:li:adTargetingFacet:staffCountRanges" + +# Brief company-size buckets -> LinkedIn staffCountRange URNs. In targetingCriteria +# the value is a range tuple URN (min,max), NOT the SIZE_* enum the facet listing +# returns; the open-ended top bucket uses INT_MAX (2147483647) as its upper bound. +COMPANY_SIZE_TO_STAFF_RANGE: dict[str, str] = { + "1": "urn:li:staffCountRange:(1,1)", + "2-10": "urn:li:staffCountRange:(2,10)", + "11-50": "urn:li:staffCountRange:(11,50)", + "51-200": "urn:li:staffCountRange:(51,200)", + "201-500": "urn:li:staffCountRange:(201,500)", + "501-1000": "urn:li:staffCountRange:(501,1000)", + "1001-5000": "urn:li:staffCountRange:(1001,5000)", + "5001-10000": "urn:li:staffCountRange:(5001,10000)", + "10001+": "urn:li:staffCountRange:(10001,2147483647)", +} + + +class _TargetingClient(Protocol): + async def typeahead_targeting_entities( + self, *, facet: str, query: str + ) -> list[dict[str, Any]]: ... + async def list_seniorities(self) -> list[dict[str, Any]]: ... + async def list_functions(self) -> list[dict[str, Any]]: ... + + +@dataclass +class ResolvedTargeting: + criteria: dict[str, Any] + unresolved: dict[str, list[str]] = field(default_factory=dict) + + +def _norm(value: str) -> str: + return " ".join(value.strip().lower().replace("-", " ").split()) + + +def _norm_size(value: str) -> str: + return value.strip().lower().replace(" ", "") + + +def _localized_name(entity: dict[str, Any]) -> str | None: + return entity.get("name", {}).get("localized", {}).get("en_US") + + +def _best_typeahead_match(query: str, hits: list[dict[str, Any]]) -> str | None: + """Prefer an exact (normalized) name match; else LinkedIn's top-ranked hit. + + Returns None only when typeahead found nothing — we never fabricate a URN. + """ + if not hits: + return None + target = _norm(query) + for hit in hits: + if _norm(hit.get("name", "")) == target: + return hit.get("urn") + return hits[0].get("urn") + + +class TargetingResolver: + """Resolves an `Audience` into a LinkedIn `targetingCriteria` payload. + + Standardized lists are fetched lazily and cached for the resolver's lifetime + (one publish), so repeated facets cost a single round-trip each. + """ + + def __init__(self, client: _TargetingClient) -> None: + self._client = client + self._seniority_index: dict[str, str] | None = None + self._function_index: dict[str, str] | None = None + + async def _seniorities(self) -> dict[str, str]: + if self._seniority_index is None: + self._seniority_index = { + _norm(name): f"urn:li:seniority:{e['id']}" + for e in await self._client.list_seniorities() + if (name := _localized_name(e)) + } + return self._seniority_index + + async def _functions(self) -> dict[str, str]: + if self._function_index is None: + self._function_index = { + _norm(name): f"urn:li:function:{e['id']}" + for e in await self._client.list_functions() + if (name := _localized_name(e)) + } + return self._function_index + + @staticmethod + def _resolve_against_index( + names: list[str], index: dict[str, str] + ) -> tuple[list[str], list[str]]: + urns: list[str] = [] + unresolved: list[str] = [] + for name in names: + urn = index.get(_norm(name)) + if urn: + urns.append(urn) + else: + unresolved.append(name) + return urns, unresolved + + async def _resolve_typeahead( + self, facet: str, names: list[str] + ) -> tuple[list[str], list[str]]: + urns: list[str] = [] + unresolved: list[str] = [] + for name in names: + hits = await self._client.typeahead_targeting_entities(facet=facet, query=name) + urn = _best_typeahead_match(name, hits) + if urn: + urns.append(urn) + else: + unresolved.append(name) + return urns, unresolved + + @staticmethod + def _resolve_company_sizes(sizes: list[str]) -> tuple[list[str], list[str]]: + values: list[str] = [] + unresolved: list[str] = [] + for size in sizes: + value = COMPANY_SIZE_TO_STAFF_RANGE.get(_norm_size(size)) + if value: + values.append(value) + else: + unresolved.append(size) + return values, unresolved + + @staticmethod + def _geo_urns(audience: Audience) -> list[str]: + urns = [ + ISO_TO_LINKEDIN_GEO_URN[code.upper()] + for code in audience.geos + if code.upper() in ISO_TO_LINKEDIN_GEO_URN + ] + # LinkedIn requires at least one location. + return urns or [ISO_TO_LINKEDIN_GEO_URN["US"]] + + async def resolve(self, audience: Audience) -> ResolvedTargeting: + clauses: list[dict[str, Any]] = [ + {"or": {FACET_LOCATIONS: self._geo_urns(audience)}} + ] + unresolved: dict[str, list[str]] = {} + + def _add(facet: str, urns: list[str], missing: list[str], key: str) -> None: + if urns: + clauses.append({"or": {facet: urns}}) + if missing: + unresolved[key] = missing + + if audience.seniorities: + urns, missing = self._resolve_against_index( + audience.seniorities, await self._seniorities() + ) + _add(FACET_SENIORITIES, urns, missing, "seniorities") + + if audience.job_functions: + urns, missing = self._resolve_against_index( + audience.job_functions, await self._functions() + ) + _add(FACET_JOB_FUNCTIONS, urns, missing, "job_functions") + + if audience.industries: + urns, missing = await self._resolve_typeahead(FACET_INDUSTRIES, audience.industries) + _add(FACET_INDUSTRIES, urns, missing, "industries") + + if audience.job_titles: + urns, missing = await self._resolve_typeahead(FACET_TITLES, audience.job_titles) + _add(FACET_TITLES, urns, missing, "job_titles") + + if audience.skills: + urns, missing = await self._resolve_typeahead(FACET_SKILLS, audience.skills) + _add(FACET_SKILLS, urns, missing, "skills") + + if audience.company_sizes: + values, missing = self._resolve_company_sizes(audience.company_sizes) + _add(FACET_STAFF_COUNT, values, missing, "company_sizes") + + return ResolvedTargeting(criteria={"include": {"and": clauses}}, unresolved=unresolved) + + +__all__ = [ + "COMPANY_SIZE_TO_STAFF_RANGE", + "FACET_INDUSTRIES", + "FACET_JOB_FUNCTIONS", + "FACET_LOCATIONS", + "FACET_SENIORITIES", + "FACET_SKILLS", + "FACET_STAFF_COUNT", + "FACET_TITLES", + "ResolvedTargeting", + "TargetingResolver", +] diff --git a/tests/integrations/test_linkedin_targeting.py b/tests/integrations/test_linkedin_targeting.py new file mode 100644 index 0000000..c15ba26 --- /dev/null +++ b/tests/integrations/test_linkedin_targeting.py @@ -0,0 +1,146 @@ +"""Tests for the B2B targeting resolver. + +These pin the two resolution paths (standardized-enum lookup vs. typeahead), +the no-guess contract (unmatched names are surfaced, never fabricated into a +URN), the targetingCriteria shape, and that standardized lists are fetched once. +""" + +from __future__ import annotations + +from yieldagent.domain import Audience +from yieldagent.integrations.linkedin.targeting import ( + FACET_INDUSTRIES, + FACET_JOB_FUNCTIONS, + FACET_LOCATIONS, + FACET_SENIORITIES, + FACET_SKILLS, + FACET_STAFF_COUNT, + FACET_TITLES, + TargetingResolver, +) + +_US_GEO = "urn:li:geo:103644278" + + +def _named(id_: int, name: str) -> dict: + return {"id": id_, "name": {"localized": {"en_US": name}}} + + +class _FakeTargetingClient: + """Serves canned standardized lists + typeahead hits; counts list fetches.""" + + def __init__(self) -> None: + self.seniority_calls = 0 + self.function_calls = 0 + self.typeahead_calls: list[tuple[str, str]] = [] + # facet -> query(lowercased) -> hits + self.typeahead: dict[str, dict[str, list[dict]]] = { + FACET_INDUSTRIES: { + "advertising": [{"urn": "urn:li:industry:80", "name": "Advertising Services"}], + }, + FACET_TITLES: { + "marketing manager": [ + {"urn": "urn:li:title:99", "name": "Senior Marketing Manager"}, + {"urn": "urn:li:title:26", "name": "Marketing Manager"}, + ], + "growth hacker": [{"urn": "urn:li:title:500", "name": "Growth Lead"}], + }, + FACET_SKILLS: { + "programmatic advertising": [ + {"urn": "urn:li:skill:60778", "name": "Programmatic Advertising"} + ], + }, + } + + async def list_seniorities(self) -> list[dict]: + self.seniority_calls += 1 + return [_named(6, "Director"), _named(7, "VP"), _named(8, "CXO")] + + async def list_functions(self) -> list[dict]: + self.function_calls += 1 + return [_named(15, "Marketing"), _named(25, "Sales")] + + async def typeahead_targeting_entities(self, *, facet: str, query: str) -> list[dict]: + self.typeahead_calls.append((facet, query)) + return self.typeahead.get(facet, {}).get(query.strip().lower(), []) + + +def _clause_facets(criteria: dict) -> dict[str, list]: + """Flatten the include/and/or structure into {facetUrn: values}.""" + out: dict[str, list] = {} + for clause in criteria["include"]["and"]: + out.update(clause["or"]) + return out + + +async def test_geo_only_defaults_to_us_when_empty() -> None: + resolver = TargetingResolver(_FakeTargetingClient()) + resolved = await resolver.resolve(Audience(description="x", geos=[])) + assert _clause_facets(resolved.criteria) == {FACET_LOCATIONS: [_US_GEO]} + assert resolved.unresolved == {} + + +async def test_enum_facets_resolve_and_surface_misses() -> None: + resolver = TargetingResolver(_FakeTargetingClient()) + resolved = await resolver.resolve( + Audience( + description="x", + geos=["US"], + seniorities=["Director", "VP", "Wizard"], # Wizard has no match + job_functions=["marketing"], # case-insensitive + ) + ) + facets = _clause_facets(resolved.criteria) + assert facets[FACET_SENIORITIES] == ["urn:li:seniority:6", "urn:li:seniority:7"] + assert facets[FACET_JOB_FUNCTIONS] == ["urn:li:function:15"] + assert resolved.unresolved == {"seniorities": ["Wizard"]} + + +async def test_typeahead_prefers_exact_then_top_hit() -> None: + resolver = TargetingResolver(_FakeTargetingClient()) + resolved = await resolver.resolve( + Audience( + description="x", + geos=["US"], + industries=["advertising"], + job_titles=["Marketing Manager", "Growth Hacker"], + skills=["Programmatic Advertising"], + ) + ) + facets = _clause_facets(resolved.criteria) + assert facets[FACET_INDUSTRIES] == ["urn:li:industry:80"] + # "Marketing Manager" exact match wins over the higher-ranked "Senior..."; the + # non-exact "Growth Hacker" falls back to LinkedIn's top hit (Growth Lead). + assert facets[FACET_TITLES] == ["urn:li:title:26", "urn:li:title:500"] + assert facets[FACET_SKILLS] == ["urn:li:skill:60778"] + assert resolved.unresolved == {} + + +async def test_typeahead_no_hits_is_unresolved_not_guessed() -> None: + resolver = TargetingResolver(_FakeTargetingClient()) + resolved = await resolver.resolve( + Audience(description="x", geos=["US"], industries=["Nonexistent Industry"]) + ) + assert FACET_INDUSTRIES not in _clause_facets(resolved.criteria) + assert resolved.unresolved == {"industries": ["Nonexistent Industry"]} + + +async def test_company_sizes_map_to_staff_ranges() -> None: + resolver = TargetingResolver(_FakeTargetingClient()) + resolved = await resolver.resolve( + Audience(description="x", geos=["US"], company_sizes=["11-50", "1001-5000", "bogus"]) + ) + facets = _clause_facets(resolved.criteria) + assert facets[FACET_STAFF_COUNT] == [ + "urn:li:staffCountRange:(11,50)", + "urn:li:staffCountRange:(1001,5000)", + ] + assert resolved.unresolved == {"company_sizes": ["bogus"]} + + +async def test_standardized_lists_fetched_once() -> None: + client = _FakeTargetingClient() + resolver = TargetingResolver(client) + await resolver.resolve(Audience(description="x", geos=["US"], seniorities=["Director", "VP"])) + await resolver.resolve(Audience(description="x", geos=["US"], seniorities=["CXO"])) + assert client.seniority_calls == 1