Building Agent Workflows

pip install trendsagi

from trendsagi import TrendsAGIClient, APIError
import os

# Initialize the sensory layer
client = TrendsAGIClient(api_key=os.getenv("TRENDSAGI_API_KEY"))

try:
    # Agent checks for high-velocity signals in the 'Technology' sector
    signals = client.get_trends(limit=5, category="Technology", sort_by="velocity")
    
    print(f"Agent Awareness Initialized. Detected {len(signals.trends)} active signals.")
    for signal in signals.trends:
        print(f" - Signal Detected: {signal.name} (Velocity: {signal.volume})")
        
except APIError as e:
    print(f"Sensory Input Failed ({e.status_code}): {e.error_detail}")

# Example: Preparing a prompt for an Ad Generator Agent
trend_id = 123 
insights = client.get_ai_insights(trend_id=trend_id)

# Constructing the System Prompt Context
# Note: These are retrieved from the cache. New insights must be triggered via the dashboard.
system_context = {
    "topic": insights.trend_name,
    "current_public_mood": insights.sentiment_summary,
    "marketing_hook": insights.suggested_marketing_angle,
    "taboo_subjects": insights.negative_themes # Guardrails
}

prompt = f"""
You are an expert copywriter. 
Topic: {system_context['topic']}
Mood: {system_context['current_public_mood']}
Hook: {system_context['marketing_hook']}

Task: Write 3 Facebook Ad headlines that align with this mood.
"""

# agent.llm.predict(prompt)

signals = client.get_trends(category="Technology", sort_by="velocity", limit=5)
for signal in signals.trends:
    insight = client.get_ai_insights(trend_id=signal.id)
    print(signal.name, insight.sentiment_summary)

1) GET /api/intelligence/recommendations
2) GET /api/insights/search
3) Compare with your search analytics metrics
4) Generate briefs for highest-overlap opportunities

# Fetch grounded, cached context before generation
trends_response = client.get_trends(search="sustainable fashion", limit=1)
trend = trends_response.trends[0]
insights = client.get_ai_insights(trend_id=trend.id)

print(f"--- Fact Sheet for Agent ---")
print(f"Context: {insights.trend_name}")
print(f"Verified Themes: {', '.join(insights.key_themes)}")
print(f"Audience Sentiment: {insights.sentiment_category}")

# The agent uses 'Verified Themes' as a checklist for content sections.

# Background Job: Update 'Trending Now' Collection

CATEGORIES = ["Fashion", "Electronics", "Home"]

for cat in CATEGORIES:
    # 1. Agent observes the market
    trends = client.get_trends(category=cat, limit=3)
    
    # 2. Agent decides which products match
    for trend in trends.trends:
        print(f"Market Signal: {trend.name}")
        # vector_search_products(query=trend.name) -> returns matching SKUs
        
    # 3. Agent updates the database
    # db.update_collection(name="trending_now", items=matching_skus)

from langchain.agents import tool
from trendsagi import TrendsAGIClient

# Initialize the Context Layer
trends_client = TrendsAGIClient(api_key="YOUR_KEY")

@tool
def check_market_signals(sector: str) -> str:
    """
    Useful for when you need to know what is currently trending or 
    popular in a specific market sector. Returns real-time data.
    """
    try:
        response = trends_client.get_trends(category=sector, limit=5)
        return str([t.name for t in response.trends])
    except Exception:
        return "Data unavailable."

# The LLM now has a "sense" of the market.
# agent_chain = initialize_agent(tools=[check_market_signals], ...)

# 1. Create a dedicated knowledge base
project = client.create_context_project(name="Support Bot Knowledge")

# 2. Upload the official policy document (PDF/Text/Image)
client.upload_context_file(
    project_id=project.id,
    file_path="./refund_policy_2025.pdf",
    item_type="reference_doc"
)

# 3. Agent Runtime: Retrieve context relevant to user query
user_query = "Can I get a refund after 30 days?"
relevant_items = client.query_context(
    project_id=project.id,
    search=user_query
)

# 4. Inject into LLM
context_block = "
".join([item.content for item in relevant_items])
# response = agent.predict(f"Context: {context_block}
Question: {user_query}")

exports.trendsReactor = async (req, res) => {
    // 1. Verify Signature (Security)
    const signature = req.headers['x-webhook-signature'];
    if (!verifySignature(req.body, signature)) {
        return res.status(401).send('Unauthorized');
    }

    // 2. Parse Payload
    const { trend_name, velocity, sentiment } = req.body;

    // 3. Execute Business Logic
    console.log(`Reacting to trend: ${trend_name} (Velocity: ${velocity})`);
    await triggerMarketingCampaign(trend_name);

    res.status(200).send('Reactor Executed');
};

import asyncio
import websockets
import json

URI = f"wss://api.trendsagi.com/ws/industry-live?token={API_KEY}"

async def run_agent_listener():
    async with websockets.connect(URI) as websocket:
        print("--> Agent Connected to Nervous System")
        while True:
            message = await websocket.recv()
            data = json.loads(message)
            
            # Immediate Reaction
            if data['type'] == 'new_market_event' and data['payload']['impact'] == 'High':
                await trigger_response_protocol(data['payload'])

# asyncio.run(run_agent_listener())

wss://api.trendsagi.com/ws/trends-live?token=KEY&trends=AI,Technology