For ADS Subscribers

npm install @agentdatashuttle/adsjs @langchain/google-genai @langchain/langgraph @langchain/core

pip install agentdatashuttle_adspy langchain-google-genai langgraph python-dotenv

import { ChatGoogleGenerativeAI } from "@langchain/google-genai";
import { HumanMessage } from "@langchain/core/messages";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import dotenv from "dotenv";
 
dotenv.config();
// Import ADS Subscriber and add ADS Data Connectors
import {
  types,
  dataconnector,
  subscriber,
  notifications,
} from "@agentdatashuttle/adsjs";
 
dotenv.config();
 
(async () => {
  // Define support ticket categorization tool
  //@ts-ignore
  const categorizeTicket = tool(
    async (args: any) => {
      // Simulate categorization logic
      const categories = ["Technical", "Billing", "General Inquiry"];
      const category =
        categories[Math.floor(Math.random() * categories.length)];
      return `Ticket categorized as: ${category}`;
    },
    {
      name: "categorize_ticket",
      description: "Categorizes a support ticket based on its content.",
      schema: z.object({
        ticket_content: z
          .string()
          .describe("The content of the support ticket."),
      }),
    }
  );
 
  // Define the tools for the agent to use
  const agentTools = [categorizeTicket];
  const llm = new ChatGoogleGenerativeAI({ model: "gemini-2.0-flash" });
 
  //@ts-ignore
  const agent = createReactAgent({
    llm: llm,
    tools: agentTools,
  });
 
  // Step 1: Define callback function for ADS Subscriber to invoke agent
  const invoke_agent = async (
    prompt: string,
    payload: types.ADSDataPayload
  ) => {
    console.log("The payload was:", payload);
 
    // Filter specific events in/out as you desire
    if (payload.event_name === "support_ticket_closed") {
      return "NO INVOCATION FOR THIS EVENT - TICKET CLOSED";
    }
 
    // Invoke your agent with the context enriched prompt generated by Agent Data Shuttle
    const response = await agent.invoke({
      messages: [new HumanMessage(prompt)],
    });
 
    // Return final agent response - will be sent to all notification channels for later review
    return response.messages[response.messages.length - 1].content as string;
  };
 
  // Step 2: Define ADSBridgeClientParams and corresponding ADSDataConnector
  const adsBridgeClientParams: types.ADSBridgeClientParams = {
    connection_string: process.env.ADS_BRIDGE_URL || "http://localhost:9999",
    path_prefix: "/ads_bridge",
    ads_subscribers_pool_id:
      process.env.SUBSCRIBER_POOL_ID || "<a_random_uuid>", // Replace with your actual pool ID to group horizontally scaled replicas of ADS Subscribers - use https://agentdatashuttle.knowyours.co/pool-id-generator to make one if needed
  };
 
  const dataConnectorOne = new dataconnector.ADSDataConnector(
    "SupportTicketConnector",
    adsBridgeClientParams
  );
 
  const redisParams: types.RedisParams = {
    host: process.env.REDIS_HOST || "localhost",
    port: parseInt(process.env.REDIS_PORT || "6379"),
    username: process.env.REDIS_USERNAME || "",
    password: process.env.REDIS_PASSWORD || "",
  };
 
  const agentDescription =
    "This agent categorizes support tickets based on their content.";
 
  // Step 3: Optionally, add notification channels
  const emailChannel = new notifications.EmailNotificationChannel(
    agentDescription,
    process.env.SMTP_HOST || "smtp.gmail.com",
    parseInt(process.env.SMTP_PORT ?? "") || 587,
    process.env.SMTP_USERNAME || "",
    process.env.SMTP_PASSWORD || "",
    process.env.FROM_EMAIL || "[email protected]",
    process.env.TO_EMAIL || "[email protected]"
  );
 
  // Step 4: Create the ADSSubscriber with the callback function, LLM, and Data Connectors.
  // The ADSSubscriber will listen for events from all the data connectors and invoke the agent.
  const adsSubscriber = new subscriber.ADSSubscriber(
    null,
    invoke_agent,
    llm,
    agentDescription,
    [dataConnectorOne],
    redisParams,
    [emailChannel]
  );
 
  // Step 5: Start the ADSSubscriber to listen for events and invoke the agent.
  await adsSubscriber.start();
})();

  import os
  from dotenv import load_dotenv
  from langchain_google_genai import ChatGoogleGenerativeAI
  from langchain_core.messages import HumanMessage
  from langgraph.prebuilt import create_react_agent
 
  from langchain_core.tools import tool
 
  # Import ADS Subscriber and add ADS Data Connectors
  from agentdatashuttle_adspy.core.subscriber import ADSSubscriber
  from agentdatashuttle_adspy.core.dataconnector import ADSDataConnector
  from agentdatashuttle_adspy.core.client import ADSBridgeClientParams
  from agentdatashuttle_adspy.models.models import ADSDataPayload
  from agentdatashuttle_adspy.core.notifications import EmailNotificationChannel
 
  # Load environment variables from .env file
  load_dotenv()
 
 
  # Define support ticket processing tools
  @tool
  def categorize_ticket(subject: str, description: str) -> str:
      """Categorizes a support ticket based on subject and description."""
      text = f"{subject} {description}".lower()
      if "payment" in text or "billing" in text:
          return "billing"
      elif "bug" in text or "error" in text:
          return "technical"
      elif "login" in text or "password" in text:
          return "account"
      else:
          return "general"
 
 
  # Define the tools for the agent to use
  agent_tools = [categorize_ticket]
  llm = ChatGoogleGenerativeAI(
      model="gemini-2.0-flash", google_api_key=os.getenv("GOOGLE_API_KEY", "")
  )
 
  agent = create_react_agent(llm, agent_tools, debug=True)
 
 
  # Step 1: Define callback function for ADS Subscriber to invoke agent
  def invoke_agent(prompt: str, payload: ADSDataPayload) -> str:
      print("The ADS payload was:", payload)
 
      # Filter specific events in/out as you desire
      if payload.event_name == "support_ticket_closed":
          return "NO AGENT INVOCATION FOR THIS EVENT - TICKET CLOSED"
 
      # Invoke your agent with the context enriched prompt generated by Agent Data Shuttle
      response = agent.invoke({"messages": [HumanMessage(prompt)]})
 
      # Return final agent response - will be sent to all notification channels for later review
      return response["messages"][-1].content
 
 
  # Step 2: Define ADSBridgeClientParams and corresponding ADSDataConnector
  ads_bridge_client_params = ADSBridgeClientParams(
      connection_string=os.getenv("ADS_BRIDGE_URL", "http://localhost:9999"),
      path_prefix="/ads_bridge",
      ads_subscribers_pool_id=os.getenv("SUBSCRIBER_POOL_ID", "<a_random_uuid>"),
  )
 
  data_connector_one = ADSDataConnector(
      connector_name="SupportTicketConnector",
      bridge_client_params=ads_bridge_client_params,
  )
 
  # Step 3: Optionally, add notification channels
  email_channel = EmailNotificationChannel(
      "AI agent that processes support tickets and provides automated responses",
      os.getenv("SMTP_HOST", "smtp.gmail.com"),
      int(os.getenv("SMTP_PORT", "587")),
      os.getenv("SMTP_USERNAME", ""),
      os.getenv("SMTP_PASSWORD", ""),
      os.getenv("FROM_EMAIL", "[email protected]"),
      os.getenv("TO_EMAIL", "[email protected]"),
  )
 
 
  # Step 4: Create the ADSSubscriber with the callback function, LLM, and Data Connectors.
  # The ADSSubscriber will listen for events from all the data connectors and invoke the agent.
  ads_subscriber = ADSSubscriber(
      agent_callback_function=invoke_agent,
      llm=llm,
      agent_description="AI agent that processes support tickets and provides automated responses",
      data_connectors=[data_connector_one],
      notification_channels=[email_channel],
  )
 
  # Step 5: Start the ADSSubscriber to listen for events and invoke the agent.
  ads_subscriber.start()