SDK Integration Prompt (Python)

System prompts for coding agents to easily integrate the Greenflash Python SDK. Copy and paste the content of this page into your coding assistant.

Using a Skills-compatible agent like Claude Code? The Greenflash agent skill handles the entire integration automatically — install, configure, and wire up logging with a single command. Run /greenflash-onboard instead of copying this prompt.

Greenflash Python SDK Integration Plan

Your task is to integrate the Greenflash Python SDK into this codebase and ship the minimal, correct implementation to start sending conversation data. Follow the steps below carefully and exactly, and make sure to use the checklist at the end to ensure you’ve followed all the steps.

Important: Before implementing, analyze the codebase to understand:

1. Is it async or sync?
2. Where does user input get received vs. where LLM responses are generated?
3. Are they in the same function/location or separate?

This analysis will determine your implementation approach (see Section 3C).

---

0) Pre-flight checks (async vs sync, aiohttp)

1. Determine whether this codebase is async:

   • Look for FastAPI/Starlette/aiohttp usage, async def route handlers, or an active asyncio loop.
   • If none of the above, treat as sync.

2. Check if aiohttp is already in use:

   • Search for import aiohttp or an aiohttp dependency in pyproject.toml / requirements.txt.

3. Install the correct SDK variant:

   • Async app (and aiohttp is present or acceptable to add):
     pip install --pre "greenflash[aiohttp]"
     
   • Sync app (or you don’t want to add aiohttp):
     pip install --pre greenflash
     

Note: The --pre flag is required because the Python SDK is currently in pre-release. The aiohttp extra provides better performance in async environments. For sync apps, the base package is sufficient.

---

1) Configuration

1. Ensure the API key is available at runtime:

   • Add to your environment (or .env if the project uses python-dotenv):
     GREENFLASH_API_KEY=YOUR_GREENFLASH_API_KEY
     
   • Do not hard-code the key.
   • The SDK will automatically read GREENFLASH_API_KEY from environment variables if not passed explicitly.

2. Ensure you have your Product ID:

   • Find your Product ID in Greenflash for the product you want to track.
   • You will pass this directly to each API call (not as an env var), since a single codebase may serve multiple products.

---

2) Create a single reusable Greenflash client

Best practice: create one module that exports a ready-to-use client. Name it greenflash_client.py. Keep all env lookups centralized here.

For sync apps:

# greenflash_client.py
import os
from greenflash import Greenflash

# The SDK will automatically read GREENFLASH_API_KEY from environment variables.
# Explicitly passing it here allows for better error messages if it's missing.
client = Greenflash(api_key=os.environ.get("GREENFLASH_API_KEY"))

For async apps:

# greenflash_client.py
import os
from greenflash import AsyncGreenflash

# The SDK will automatically read GREENFLASH_API_KEY from environment variables.
# Explicitly passing it here allows for better error messages if it's missing.
# Use AsyncGreenflash for native async support.
client = AsyncGreenflash(api_key=os.environ.get("GREENFLASH_API_KEY"))

Place this file in a shared module location (e.g., app/services/greenflash_client.py).

Import client wherever you log messages or identify users/orgs.

---

3) Log messages with client.messages.create (required fields only)

Required fields:

• product_id
• external_user_id
• external_conversation_id
• messages → list of { "role": "user" | "assistant" | "system", "content": str }

Recommended optional fields:

• model — the AI model used (e.g., "gpt-4o", "claude-sonnet-4-20250514"). Enables model performance comparison in Greenflash.
• external_organization_id — links conversations to an organization/account for segment analysis.
• properties — dict of custom metadata to attach to the conversation.

3A) If the app is async, use AsyncGreenflash

Use AsyncGreenflash for native async support. Fire-and-forget with asyncio.create_task() to avoid blocking.

# example_async_logging.py
import asyncio
from typing import List, Dict
from greenflash_client import client

async def log_chat_async(product_id: str, user_id: str, convo_id: str, chat: List[Dict[str, str]]) -> None:
    """
    Helper function for logging messages to Greenflash.

    chat can be:
    - A single message: [{"role": "user", "content": "Hello"}]
    - A turn: [{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi!"}]
    - Multiple turns/conversation

    See section 3C for deciding which pattern to use.
    """
    params = {
        "product_id": product_id,
        "external_user_id": user_id,
        "external_conversation_id": convo_id,
        "messages": chat,
    }
    # Fire-and-forget so LLM response isn't delayed
    # Note: client is AsyncGreenflash, so this call is async
    asyncio.create_task(client.messages.create(**params))

Note: If you’re using the sync Greenflash client in an async app, wrap the call with asyncio.to_thread() instead: asyncio.create_task(asyncio.to_thread(client.messages.create, **params))

3B) If the app is sync, call directly

# example_sync_logging.py
from typing import List, Dict
from greenflash_client import client

def log_chat_sync(product_id: str, user_id: str, convo_id: str, chat: List[Dict[str, str]]) -> None:
    """
    Helper function for logging messages to Greenflash.

    chat can be:
    - A single message: [{"role": "user", "content": "Hello"}]
    - A turn: [{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi!"}]
    - Multiple turns/conversation

    See section 3C for deciding which pattern to use.
    """
    params = {
        "product_id": product_id,
        "external_user_id": user_id,
        "external_conversation_id": convo_id,
        "messages": chat,
    }
    client.messages.create(**params)

3C) Where to call it in your flow — choosing the right pattern

The Greenflash messages API is flexible: You can send 1 to many messages per call. Subsequent calls are automatically ordered by timestamp, so you can safely send:

• Each message individually (user message, then assistant message separately)
• Messages grouped as turns (user + assistant together in one call)
• An entire conversation at once (less common for real-time logging)

How to decide which pattern to use:

1. Analyze your codebase structure first:

   • Look for where user input is received and where the assistant/LLM response is generated
   • Check if these happen in the same function or in different parts of the code

2. Pattern A - Send as turns:

   • Use if: User input and assistant output are available together in the same function/location
   • Example: Your LLM handler receives user input, generates a response, and returns both
   • Benefit: One API call per turn, simpler tracking
   # After generating the assistant response
   messages = [
       {"role": "user", "content": user_input},
       {"role": "assistant", "content": assistant_response}
   ]
   asyncio.create_task(client.messages.create(
       product_id="YOUR_PRODUCT_ID",
       external_user_id=user_id,
       external_conversation_id=convo_id,
       messages=messages
   ))
   

3. Pattern B - Send individually:

   • Use if: User input and assistant output are handled in separate functions/locations
   • Example: User input handler is separate from the LLM response generation
   • Benefit: Log as events happen, no need to refactor code structure
   # When user message is received
   asyncio.create_task(client.messages.create(
       product_id="YOUR_PRODUCT_ID",
       external_user_id=user_id,
       external_conversation_id=convo_id,
       messages=[{"role": "user", "content": user_input}]
   ))
   
   # Later, when assistant responds
   asyncio.create_task(client.messages.create(
       product_id="YOUR_PRODUCT_ID",
       external_user_id=user_id,
       external_conversation_id=convo_id,
       messages=[{"role": "assistant", "content": assistant_response}]
   ))
   

Key points:

• Greenflash automatically orders messages by timestamp across calls, so both patterns work correctly
• Choose the pattern that requires minimal changes to the existing codebase
• You can mix patterns if needed (e.g., some conversations logged as turns, others as individual messages)

---

4) Minimal examples for message payload

messages = [
  {"role": "user", "content": "What is the weather like today?"},
  {"role": "assistant", "content": "I can help you check the weather. What city are you in?"}
]

---

5) (Optional) Agentic Workflows

If the product is an AI agent with tool calls, reasoning steps, or multi-step workflows, you can use structured message_type values instead of simple role to give Greenflash richer visibility into agent behavior.

Key concepts:

• Use message_type instead of role to describe what the agent is doing
• Use content only for user-facing input/output
• Use output for internal agent data (thoughts, tool results)
• Use input for tool call parameters
• Link related messages with external_message_id and parent_external_message_id

Example: Agent with tool call

client.messages.create(
    product_id="YOUR_PRODUCT_ID",
    external_user_id=user_id,
    external_conversation_id=convo_id,
    messages=[
        # User query (user-facing)
        {"role": "user", "content": "What's my account balance?"},
        # Agent reasoning (internal)
        {
            "external_message_id": "thought-1",
            "message_type": "thought",
            "output": "User wants account balance. Need to call account_lookup tool.",
        },
        # Tool invocation
        {
            "external_message_id": "tool-1",
            "message_type": "tool_call",
            "tool_name": "account_lookup",
            "input": {"user_id": "12345"},
        },
        # Tool result (internal)
        {
            "external_message_id": "obs-1",
            "message_type": "observation",
            "parent_external_message_id": "tool-1",
            "output": {"balance": 1250.00, "currency": "USD"},
        },
        # Final response (user-facing)
        {"role": "assistant", "content": "Your current account balance is $1,250.00."},
    ],
)

For more details: See the Agentic Messages prompt for comprehensive guidance on message types, content vs output, and modeling complex agent execution.

---

6) (Optional) Identify users — only the required field

Call once when a user logs in or as soon as a stable ID is available. Required: external_user_id.

Optional fields you can include: external_organization_id, name, email, phone, properties (dict for custom data).

Async app (using AsyncGreenflash)

# identify_user_async.py
import asyncio
from greenflash_client import client

async def identify_user(user_id: str, org_id: str | None = None) -> None:
    params = {"external_user_id": user_id}
    if org_id:
        params["external_organization_id"] = org_id
    # Fire-and-forget
    asyncio.create_task(client.users.create(**params))

Sync app

# identify_user_sync.py
from greenflash_client import client

def identify_user(user_id: str, org_id: str | None = None) -> None:
    params = {"external_user_id": user_id}
    if org_id:
        params["external_organization_id"] = org_id
    client.users.create(**params)

---

7) (Optional) Identify organizations — only the required field

Use when you have a clear organization/account ID. Required: external_organization_id.

Optional fields you can include: name, properties (dict for custom data).

Async app (using AsyncGreenflash)

# identify_org_async.py
import asyncio
from greenflash_client import client

async def identify_org(org_id: str) -> None:
    params = {"external_organization_id": org_id}
    # Fire-and-forget
    asyncio.create_task(client.organizations.create(**params))

Sync app

# identify_org_sync.py
from greenflash_client import client

def identify_org(org_id: str) -> None:
    client.organizations.create(external_organization_id=org_id)

---

8) Error handling (recommended for production)

While fire-and-forget logging is appropriate for most cases, you may want to handle errors for monitoring purposes:

import greenflash

try:
    client.messages.create(
        product_id="YOUR_PRODUCT_ID",
        external_user_id="USER_ID",
        external_conversation_id="CONVO_ID",
        messages=[{"role": "user", "content": "Hello"}],
    )
except greenflash.APIConnectionError as e:
    print(f"Connection error: {e}")
except greenflash.RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
except greenflash.APIStatusError as e:
    print(f"API error {e.status_code}: {e.response}")
except greenflash.APIError as e:
    print(f"API error occurred: {e}")

Available error classes: APIConnectionError, RateLimitError, APIStatusError, APITimeoutError, APIError (base class).

---

9) File/PR checklist

• Installed the correct package:
  • Async + ok with aiohttp → pip install --pre "greenflash[aiohttp]"
  • Otherwise → pip install --pre greenflash
• Added GREENFLASH_API_KEY to runtime env (not hard-coded).
• Have your Product ID ready (from Greenflash) to pass directly in each API call.
• Created greenflash_client.py with the appropriate client:
  • Async apps → AsyncGreenflash
  • Sync apps → Greenflash
• Analyzed codebase to determine message logging pattern:
  • User input + assistant output in same location → use Pattern A (send as turns)
  • User input + assistant output in separate locations → use Pattern B (send individually)
• Logging messages with only required fields: product_id, external_user_id, external_conversation_id, messages.
• Passing product_id directly in each client.messages.create(...) call (not from an env var).
• In async paths, used asyncio.create_task(client.messages.create(...)) for fire-and-forget logging (non-blocking).
• (Optional) Added client.users.create(...) where a stable user ID exists.
• (Optional) Added client.organizations.create(...) where an org ID exists.

Deliverable: a single PR containing greenflash_client.py, minimal logging wired into the main chat flow (using the appropriate pattern based on codebase structure), and optional user/org identification hooks. Include README notes describing env vars and where the logging occurs.