Linq API Integration

from config import linq

linq.api_token          # Bearer token for Linq API
linq.phone_number       # CareSupport's Linq Blue number (E.164)
linq.base_url           # https://api.linqapp.com/api/partner/v3
linq.webhook_signing_secret  # HMAC secret for webhook verification

{
  "linq_api_token": "...",
  "linq_phone": "+16517037981",
  "base_url": "https://api.linqapp.com/api/partner/v3",
  "webhook_signing_secret": "..."
}

async def create_chat(
    to_phone: str,
    initial_message: str,
    from_phone: str = "",
    preferred_service: str | None = None,
    effect: dict | None = None
) -> dict

result = await create_chat(
    to_phone="+16517037981",
    initial_message="Welcome to CareSupport!",
    preferred_service="iMessage",
    effect={"type": "screen", "name": "balloons"}
)

if result["success"]:
    chat_id = result["chat_id"]
    print(f"Chat created: {chat_id} via {result['service']}")

async def send_message(
    chat_id: str,
    text: str,
    preferred_service: str | None = None,
    effect: dict | None = None,
    reply_to_message_id: str | None = None,
    media_url: str | None = None,
    attachment_id: str | None = None
) -> dict

result = await send_message(
    chat_id="550e8400-e29b-41d4-a716-446655440000",
    text="Liban will pick up Degitu at 8am tomorrow.",
    effect={"type": "screen", "name": "confetti"}
)

async def list_chats(
    from_phone: str = "",
    limit: int = 20,
    cursor: str = ""
) -> dict

async def start_typing(chat_id: str) -> dict

async def stop_typing(chat_id: str) -> dict

async def mark_as_read(chat_id: str) -> dict

async def add_reaction(
    message_id: str,
    reaction_type: str = "love",
    custom_emoji: str = "",
    part_index: int = 0
) -> dict

async def remove_reaction(
    message_id: str,
    reaction_type: str = "love"
) -> dict

# React with ❤️ to a message
await add_reaction(
    message_id="550e8400-e29b-41d4-a716-446655440000",
    reaction_type="love"
)

# Custom emoji reaction
await add_reaction(
    message_id="550e8400-e29b-41d4-a716-446655440000",
    reaction_type="custom",
    custom_emoji="🎉"
)

async def send_voice_memo(
    chat_id: str,
    voice_memo_url: str,
    from_phone: str = ""
) -> dict

async def share_contact_card(chat_id: str) -> dict

async def add_participant(chat_id: str, handle: str) -> dict

async def remove_participant(chat_id: str, handle: str) -> dict

async def pre_upload_attachment(
    filename: str,
    content_type: str,
    size_bytes: int
) -> dict

# 1. Pre-upload to get attachment_id
result = await pre_upload_attachment(
    filename="care_plan.pdf",
    content_type="application/pdf",
    size_bytes=5242880  # 5MB
)

if result["success"]:
    # 2. PUT file to upload_url (use requests, httpx, etc.)
    import httpx
    async with httpx.AsyncClient() as client:
        await client.put(result["upload_url"], content=pdf_bytes)
    
    # 3. Send message with attachment_id
    await send_message(
        chat_id=chat_id,
        text="Here's your care plan.",
        attachment_id=result["attachment_id"]
    )

async def list_webhook_subscriptions() -> dict

async def create_webhook_subscription(
    target_url: str,
    events: list[str],
    version: str = "2026-02-03"
) -> dict

result = await create_webhook_subscription(
    target_url="https://caresupport.app/webhooks/linq",
    events=["message.created", "reaction.created"]
)

if result["success"]:
    # CRITICAL: Store signing_secret in config
    signing_secret = result["signing_secret"]
    print(f"Subscription ID: {result['subscription_id']}")

async def delete_webhook_subscription(subscription_id: str) -> dict

def verify_webhook_signature(
    payload: bytes,
    timestamp: str,
    signature: str
) -> bool

from flask import Flask, request
from linq_gateway import verify_webhook_signature

app = Flask(__name__)

@app.route("/webhooks/linq", methods=["POST"])
def handle_webhook():
    payload = request.get_data()
    timestamp = request.headers.get("X-Linq-Timestamp")
    signature = request.headers.get("X-Linq-Signature")
    
    if not verify_webhook_signature(payload, timestamp, signature):
        return "Invalid signature", 401
    
    event = request.json
    if event["event_type"] == "message.created":
        # Process inbound message
        chat_id = event["data"]["chat_id"]
        text = event["data"]["message"]["parts"][0]["value"]
        # ...
    
    return "", 200

async def list_phone_numbers() -> dict

# Create a new chat
python runtime/scripts/linq_gateway.py create \
  --to "+16517037981" \
  --body "Welcome to CareSupport!" \
  --service iMessage

# Send to existing chat
python runtime/scripts/linq_gateway.py send \
  --chat-id "550e8400-e29b-41d4-a716-446655440000" \
  --body "Liban will drive tomorrow at 8am."

# List all chats
python runtime/scripts/linq_gateway.py list-chats

# List phone numbers
python runtime/scripts/linq_gateway.py phones

# Start typing indicator
python runtime/scripts/linq_gateway.py typing \
  --chat-id "550e8400-e29b-41d4-a716-446655440000"

# React to a message
python runtime/scripts/linq_gateway.py react \
  --message-id "550e8400-e29b-41d4-a716-446655440000" \
  --type love

# List webhook subscriptions
python runtime/scripts/linq_gateway.py webhooks

from linq_gateway import list_chats, get_messages
from sms_handler import handle_sms, resolve_chat_id

# Poll for new messages
chats = await list_chats(limit=50)
for chat in chats["chats"]:
    messages = await get_messages(chat["id"], limit=10)
    for msg in messages["messages"]:
        if msg["direction"] == "inbound" and not_yet_processed(msg["id"]):
            # Resolve chat_id to member
            member = resolve_chat_id(chat["id"])
            if member:
                # Pass to SMS handler
                result = await handle_sms(
                    from_phone=member["phone"],
                    body=msg["parts"][0]["value"],
                    service=msg["service"]  # "iMessage", "RCS", or "SMS"
                )