Publish Article to Teams Example

This page walks through a working example of a web-based publish tool built with Python and Streamlit. It covers everything needed to authenticate, load simulation data, and publish articles to one or more teams and explains the design decisions behind the code so you can adapt it confidently for your own integration.

The complete source files are at the end of this page.

Prerequisites

• Python 3.10 or later

• A Conducttr API key (your Bearer token)

• Two packages:

pip install streamlit requests

Everything else used in the code (io, json, os, re, time, zipfile) is part of the Python standard library.

How the app works

The app follows a linear five-step flow, each gated on the previous:

1

Authenticate

Enter your API key and connect.

2

Choose a persona

Select which organisation will author the article.

3

Choose team(s)

Select one or more recipient teams.

4

Compose the article

Enter title, body, sentiment, and draft toggle.

5

Publish

Review a summary, confirm, and send.

Authentication

All Eagle API requests require a Bearer token in the Authorization header. The token/key is unique to the simulation space and is available via your account team.

The app accepts the key in two ways, in order of preference:

Environment variable (recommended for production):

export CONDUCTTR_API_KEY="your-token-here"
streamlit run conducttr_app.py

If you're running your code in Streamlit from Github then the app will be streamlit_app.py

UI entry: If no environment variable is found, the app shows a password field. The key is stored in Streamlit's st.session_state for the duration of the browser session and never written to disk.

The make_headers() helper constructs the correct headers from the key:

def make_headers(api_key: str) -> dict:
    return {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

If the Connect call returns a 4xx error, the app surfaces the HTTP status and prompts you to check the key before proceeding.

---

Fetching Personas

Personas are the non-player character entities available in the exercise. Note that they have to be published to the simulation space. The Eagle API does not return persona data as direct JSON, it returns a url to a zip file of personas. Instead it works in two steps:

Step 1 — GET /v1.1/eagle/personas returns a short-lived presigned URL:

{ "presigned_url": "https://..." }

Step 2 — Fetch that URL to download a ZIP archive containing a single .json file with all persona data.

@st.cache_data(ttl=300, show_spinner=False)
def fetch_personas(api_key: str) -> list[dict]:
    res = requests.get(f"{API_BASE}/personas", headers=make_headers(api_key))
    res.raise_for_status()
    presigned_url = res.json()["presigned_url"]

    zip_res = requests.get(presigned_url)
    zip_res.raise_for_status()

    with zipfile.ZipFile(io.BytesIO(zip_res.content)) as z:
        json_filename = next(n for n in z.namelist() if n.endswith(".json"))
        data = json.loads(z.read(json_filename))

    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        for value in data.values():
            if isinstance(value, list):
                return value
    return []

The JSON structure inside the ZIP is a list of persona objects.

Filtering to organisations

The app filters to organisation personas only (system_info.is_organisation === true) because only organisations can have websites:

orgs = [p for p in personas_raw if p.get("system_info", {}).get("is_organisation")]

Key persona fields

All fields used for publishing are nested inside system_info:

Field	Path	Description
Unique identifier	system_info.hash	Required for publishing. Pass this as the persona field in the POST body
Display name	system_info.name	Human-readable name shown in the UI
Handle	system_info.handle	Short username, e.g. UNODA
Profile image	system_info.profile_image_url	Avatar URL
Is organisation	system_info.is_organisation	true / false

Important: The endpoint expects system_info.hash as the persona value. Using the wrong identifier will result in a 400 or 404 error.

Caching

Personas are set up before the exercise begins (before STARTEX) and are not expected to change during a running exercise. The app caches the persona list for 5 minutes using @st.cache_data(ttl=300). In practice you could increase this TTL significantly or cache indefinitely for the duration of a session since persona data will not (typically) change mid-exercise.

If you do need to force a reload (for example, after a pre- or mid-exercise configuration change), use the Refresh data button in the app footer, which calls fetch_personas.clear() to invalidate the cache immediately.

Fetching Teams

Teams are the player structures that receive published content in this example. GET /v1.1/eagle/teams returns a list directly:

@st.cache_data(ttl=300, show_spinner=False)
def fetch_teams(api_key: str) -> list[dict]:
    res = requests.get(f"{API_BASE}/teams", headers=make_headers(api_key))
    res.raise_for_status()
    data = res.json()

    if isinstance(data, list):
        raw = data
    elif isinstance(data, dict):
        raw = data.get("teams") or data.get("data") or []
    else:
        raise ValueError(f"Unexpected /teams response: {type(data)}")

    return [{"team_id": t.get("team_id") or t.get("id"), "name": t["name"]} for t in raw]

The response may be a bare array or wrapped in a teams or data key — the function normalises all three shapes into a consistent list of { team_id, name } objects.

Team naming conventions

Raw team names follow a convention that can be cleaned up for display:

Raw name	Displayed as	Rule
T - Syndicate A - 2025/02/15	Syndicate A	Strip T - prefix and trailing date
T - Red Cell	Red Cell	Strip T - prefix
S - Session	Session	Any S - prefix becomes "Session"

def clean_team_name(raw_name: str) -> str:
    # Named prefixes with fixed display labels
    if raw_name.startswith("S - "):
        return "Session"
    if raw_name.startswith("M - "):
        return "Moderators"
    # All other prefixes (e.g. T -): strip prefix, then strip trailing date/timestamp
    name = re.sub(r"^[A-Z] - ", "", raw_name)
    name = re.sub(r" - \d{4}[\/-]\d{2}[\/-]\d{2}.*$", "", name)
    return name.strip()

Like personas, team composition is established before STARTEX. The same 5-minute cache and manual refresh approach applies.

---

Publishing a Message

Publishing sends a POST /v1.1/eagle/messages for each selected team. The app sends one request per team with a 500ms delay between calls to stay within rate limits.

Request payload

{
  "persona":   "3-49786",
  "channel":   "websites",
  "title":     "Ceasefire agreed after overnight talks",
  "subtitle":  "Negotiators reach deal at 3am",
  "body":      "<p>Senior officials confirmed...</p>",
  "assets":    [],
  "sentiment": "positive",
  "team_id":   3442873,
  "type":      "team",
  "isDraft":   0
}

Field reference

Field	Type	Required	Notes
persona	string	Yes	system_info.hash from the personas endpoint
channel	string	Yes	Use "websites" for web article publishing
title	string	Yes	Article headline
subtitle	string	No	Standfirst / deck copy
body	string	Yes	HTML supported. Plain text is auto-wrapped in <p> tags
assets	array	No	Image or video objects — see below
sentiment	string	Yes	"positive", "neutral", or "negative"
team_id	integer	Yes	From the teams endpoint
type	string	Yes	Always "team"
isDraft	integer	Yes	0 = publish immediately, 1 = save as draft

Asset objects

To attach an image or video, include it in the assets array:

{
  "type":     "image",
  "assetUrl": "https://example.com/photo.jpg",
  "alt":      "Caption text"
}

The assetUrl must be a publicly accessible URL. The current app passes an empty assets array — extend the compose form if you need media attachment support.

Publishing to multiple teams

The app loops through selected teams and posts one request per team:

for i, team_id in enumerate(team_ids):
    payload = { ... "team_id": team_id ... }
    result = publish_to_team(api_key, payload)
    results.append({"team_id": team_id, **result})
    if i < len(team_ids) - 1:
        time.sleep(0.5)   # 500ms between requests

Results are collected and displayed per team so any individual failure is immediately visible without affecting the other publishes.

---

Error handling

publish_to_team() catches both HTTP errors and network exceptions without raising, returning a consistent result dict:

def publish_to_team(api_key: str, payload: dict) -> dict:
    try:
        res = requests.post(f"{API_BASE}/messages", headers=make_headers(api_key), json=payload)
        if res.ok:
            return {"ok": True, "status": res.status_code}
        return {"ok": False, "status": res.status_code, "error": res.text[:300]}
    except requests.RequestException as e:
        return {"ok": False, "status": 0, "error": str(e)}

HTTP status codes

Status	Meaning	Action
200	Success	—
400	Bad request	Check payload — likely a missing required field or wrong persona hash
401	Unauthorised	API key is missing or invalid
403	Forbidden	API key does not have permission to publish
404	Not found	Invalid endpoint, or team_id does not exist
429	Rate limited	Reduce request frequency; increase the sleep delay
500	Server error	Retry after a short wait

---

Source files

conducttr_app.py

"""
Conducttr Eagle API — Streamlit Publish App
Usage:
    pip install streamlit requests
    streamlit run conducttr_app.py
"""

import io, json, os, re, time, zipfile
import requests
import streamlit as st

# ── Page config ────────────────────────────────────────────────────────────────

st.set_page_config(
    page_title="Conducttr Publisher",
    page_icon="📡",
    layout="centered",
)

# ── Constants ──────────────────────────────────────────────────────────────────

API_BASE = "https://api.conducttr.com/v1.1/eagle"

# ── Helpers ────────────────────────────────────────────────────────────────────

def make_headers(api_key: str) -> dict:
    return {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }


def clean_team_name(raw_name: str) -> str:
    # Named prefixes with fixed display labels
    if raw_name.startswith("S - "):
        return "Session"
    if raw_name.startswith("M - "):
        return "Moderators"
    # All other prefixes (e.g. T -): strip prefix, then strip trailing date/timestamp
    name = re.sub(r"^[A-Z] - ", "", raw_name)
    name = re.sub(r" - \d{4}[\/-]\d{2}[\/-]\d{2}.*$", "", name)
    return name.strip()

# ── Cached API calls ───────────────────────────────────────────────────────────

@st.cache_data(ttl=300, show_spinner=False)
def fetch_personas(api_key: str) -> list[dict]:
    """Fetch all personas; cached for 5 minutes."""
    headers = make_headers(api_key)
    res = requests.get(f"{API_BASE}/personas", headers=headers)
    res.raise_for_status()
    presigned_url = res.json()["presigned_url"]

    zip_res = requests.get(presigned_url)
    zip_res.raise_for_status()

    with zipfile.ZipFile(io.BytesIO(zip_res.content)) as z:
        json_filename = next(n for n in z.namelist() if n.endswith(".json"))
        data = json.loads(z.read(json_filename))

    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        for value in data.values():
            if isinstance(value, list):
                return value
    return []


@st.cache_data(ttl=300, show_spinner=False)
def fetch_teams(api_key: str) -> list[dict]:
    """Fetch all teams; cached for 5 minutes.
    Response is always a bare JSON array with 'id' and 'name' fields.
    """
    headers = make_headers(api_key)
    res = requests.get(f"{API_BASE}/teams", headers=headers)
    res.raise_for_status()
    return [{"team_id": t["id"], "name": t["name"]} for t in res.json()]


def publish_to_team(api_key: str, payload: dict) -> dict:
    """POST a single message. Returns result dict."""
    try:
        res = requests.post(f"{API_BASE}/messages", headers=make_headers(api_key), json=payload)
        if res.ok:
            return {"ok": True, "status": res.status_code}
        return {"ok": False, "status": res.status_code, "error": res.text[:300]}
    except requests.RequestException as e:
        return {"ok": False, "status": 0, "error": str(e)}

# ── UI ─────────────────────────────────────────────────────────────────────────

st.title("📡 Conducttr Publisher")
st.caption("Publish articles to your Conducttr simulation via the Eagle API.")

# ── Step 1: API key ────────────────────────────────────────────────────────────

with st.expander("🔑 API Key", expanded="api_key" not in st.session_state):
    default_key = os.environ.get("CONDUCTTR_API_KEY", "")
    api_key_input = st.text_input(
        "Conducttr API Key",
        value=st.session_state.get("api_key", default_key),
        type="password",
        placeholder="Paste your Bearer token here",
    )
    if st.button("Connect", type="primary"):
        if not api_key_input.strip():
            st.error("Please enter an API key.")
        else:
            with st.spinner("Connecting and loading data..."):
                try:
                    personas_raw = fetch_personas(api_key_input.strip())
                    teams_raw    = fetch_teams(api_key_input.strip())
                    orgs = [p for p in personas_raw if p.get("system_info", {}).get("is_organisation")]
                    st.session_state["api_key"] = api_key_input.strip()
                    st.session_state["orgs"]    = orgs
                    st.session_state["teams"]   = teams_raw
                    st.success(f"Connected — {len(orgs)} persona(s), {len(teams_raw)} team(s) loaded.")
                    st.rerun()
                except requests.HTTPError as e:
                    st.error(f"API error: {e.response.status_code} — check your key and try again.")
                except Exception as e:
                    st.error(f"Unexpected error: {e}")

# ── Only show the rest once connected ─────────────────────────────────────────

if "api_key" not in st.session_state:
    st.info("Enter your API key above to get started.")
    st.stop()

api_key = st.session_state["api_key"]
orgs    = st.session_state["orgs"]
teams   = st.session_state["teams"]

if not orgs:
    st.error("No organisation personas found. Cannot publish.")
    st.stop()

if not teams:
    st.error("No teams found. Cannot publish.")
    st.stop()

# ── Step 2: Persona ────────────────────────────────────────────────────────────

st.divider()
st.subheader("1 · Choose a Persona")

persona_options = {
    f"{p.get('system_info', {}).get('name', 'Unknown')}  (@{p.get('system_info', {}).get('handle', '?')})": p
    for p in orgs
}
chosen_persona_label = st.selectbox(
    "Publish as",
    options=list(persona_options.keys()),
    help="Organisation personas only. Persona list refreshes every 5 minutes.",
)
chosen_persona = persona_options[chosen_persona_label]

# ── Step 3: Teams ──────────────────────────────────────────────────────────────

st.divider()
st.subheader("2 · Choose Team(s)")

team_options = {
    f"{clean_team_name(t['name'])}  (id: {t['team_id']})": t
    for t in teams
}
chosen_team_labels = st.multiselect(
    "Publish to",
    options=list(team_options.keys()),
    placeholder="Select one or more teams…",
    help="Hold Ctrl / Cmd to select multiple. Team list refreshes every 5 minutes.",
)

col_all, col_clear = st.columns([1, 5])
with col_all:
    if st.button("Select all"):
        # Trigger a rerun with all teams pre-selected via query param workaround
        st.session_state["select_all_teams"] = True
        st.rerun()

# Handle "select all" on rerun
if st.session_state.pop("select_all_teams", False):
    chosen_team_labels = list(team_options.keys())

chosen_teams = [team_options[label] for label in chosen_team_labels]

# ── Step 4: Article ────────────────────────────────────────────────────────────

st.divider()
st.subheader("3 · Compose Article")

title    = st.text_input("Title *", placeholder="Breaking: Major event unfolds…")
subtitle = st.text_input("Subtitle", placeholder="Optional standfirst or deck copy")
body_raw = st.text_area(
    "Body *",
    height=200,
    placeholder="Paste plain text or HTML. Plain text will be wrapped in <p> tags automatically.",
)
body = body_raw.strip() if "<" in body_raw else f"<p>{body_raw.strip()}</p>"

col_sent, col_draft = st.columns(2)
with col_sent:
    sentiment = st.selectbox("Sentiment", ["positive", "neutral", "negative"])
with col_draft:
    is_draft = st.checkbox("Save as draft (don't publish yet)")

# ── Step 5: Publish ────────────────────────────────────────────────────────────

st.divider()

# Validation
ready = title.strip() and body_raw.strip() and chosen_teams

if not title.strip():
    st.warning("A title is required.")
if not body_raw.strip():
    st.warning("A body is required.")
if not chosen_teams:
    st.warning("Select at least one team.")

# Summary before submitting
if ready:
    with st.expander("📋 Review before publishing", expanded=True):
        st.markdown(f"**Persona:** {chosen_persona_label}")
        st.markdown(f"**Teams:** {', '.join(clean_team_name(t['name']) for t in chosen_teams)}")
        st.markdown(f"**Title:** {title}")
        st.markdown(f"**Sentiment:** {sentiment}")
        st.markdown(f"**Mode:** {'🗒️ Draft' if is_draft else '🚀 Publish immediately'}")

publish_btn = st.button(
    "🚀 Publish" if not is_draft else "💾 Save Draft",
    type="primary",
    disabled=not ready,
    use_container_width=True,
)

if publish_btn and ready:
    article = {
        "title":     title.strip(),
        "subtitle":  subtitle.strip(),
        "body":      body,
        "sentiment": sentiment,
        "is_draft":  1 if is_draft else 0,
    }
    persona_hash = chosen_persona["system_info"]["hash"]
    team_ids     = [t["team_id"] for t in chosen_teams]

    results = []
    progress = st.progress(0, text="Publishing…")

    for i, team_id in enumerate(team_ids):
        payload = {
            "persona":   persona_hash,
            "channel":   "websites",
            "title":     article["title"],
            "subtitle":  article["subtitle"],
            "body":      article["body"],
            "assets":    [],
            "sentiment": article["sentiment"],
            "team_id":   team_id,
            "type":      "team",
            "isDraft":   article["is_draft"],
        }
        result = publish_to_team(api_key, payload)
        results.append({"team_id": team_id, **result})
        progress.progress((i + 1) / len(team_ids), text=f"Published {i + 1} of {len(team_ids)}…")
        if i < len(team_ids) - 1:
            time.sleep(0.5)

    progress.empty()

    # Results
    st.divider()
    st.subheader("Results")
    successes = [r for r in results if r["ok"]]
    failures  = [r for r in results if not r["ok"]]

    if successes:
        st.success(f"✓ Published to {len(successes)} team(s) successfully.")
    if failures:
        st.error(f"✗ {len(failures)} publish(es) failed.")

    for r in results:
        team_name = next(
            (clean_team_name(t["name"]) for t in teams if t["team_id"] == r["team_id"]),
            str(r["team_id"]),
        )
        if r["ok"]:
            st.markdown(f"✅ **{team_name}** — HTTP {r['status']}")
        else:
            st.markdown(f"❌ **{team_name}** — HTTP {r['status']}: `{r.get('error', 'unknown error')}`")

# ── Footer ─────────────────────────────────────────────────────────────────────

st.divider()
col1, col2 = st.columns([3, 1])
with col1:
    st.caption(f"Connected to `{API_BASE}`")
with col2:
    if st.button("🔄 Refresh data"):
        fetch_personas.clear()
        fetch_teams.clear()
        st.session_state.pop("orgs", None)
        st.session_state.pop("teams", None)
        st.session_state.pop("api_key", None)
        st.rerun()

requirements.txt

requests==2.32.5
streamlit==1.54.0