# Agent
Source: https://docs.superagentx.ai/agent
An `Agent` from `superagentx.agent` is a system or program that is initialized with specific goals and roles, enabling it to interact with a
large language model (LLM) and use a predefined prompt template. It is designed to perform agent by following a
set of instructions and can be customized with various parameters to adapt its behavior to different agent or
workflows. Essentially, the `agent` acts as a flexible tool that can be fine-tuned to carry out specific functions
based on the needs of the user.
### Agent Parameters
| Attribute | Parameters | Description |
| :----------------------------- | :---------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Goal** | `goal` | The primary objective or goal that the engine is designed to achieve. |
| **Role** | `role` | The role or function that the engine will assume in its operations. |
| **LLM Client** | `llm` | Interface for communicating with the large language model (LLM). |
| **Prompt Template** | `prompt_template` | Defines the structure and format of prompts sent to the LLM using PromptTemplate. |
| **Engines** | `engines` | A list of engines (or lists of engines) that the engine can utilize. This allows for flexibility in processing and task execution based on different capabilities or configurations. |
| **Agent ID** *(optional)* | `agent_id` | A unique identifier for the engine. If not provided, a new UUID will be generated by default. Useful for tracking or referencing the engine in multi-engine environments. |
| **Name** *(optional)* | `name` | An optional name for the engine, providing a more friendly reference for display or logging purposes. |
| **Description** *(optional)* | `description` | An optional description that provides additional context or details about the engine's purpose and capabilities. |
| **Output Format** *(optional)* | `output_format` | Specifies the desired format for the engine's output. This can dictate how results are structured and presented. |
| **Max Retry** *(optional)* | `max_retry` | The maximum number of retry attempts for operations that may fail.Default is set to 5. This is particularly useful in scenarios where transient errors may occur, ensuring robust execution. |
```python theme={null}
from superagentx.agent import Agent
ecom_agent = Agent(
name='Ecom Agent',
goal="Get me the best search results",
role="You are the best product searcher",
llm=llm_client,
prompt_template=prompt_template,
engines=[[amazon_engine, walmart_engine]],
agent_id="ecom-id-1",
description="""E-commerce enables buying and selling products online,
providing customers with the convenience of shopping anytime, anywhere.
It connects buyers and sellers through digital platforms, offering
secure transactions and delivery.
""",
output_format="Generate as JSON format",
max_retry=5
)
```
## Agent Configuration
### Sequence
In sequence execution, `engine(s)` are performed one after another, with each `engine(s)` waiting for the
previous one to finish before starting. This approach is simple but can be slow if `engine(s)` are independent.
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...
engines=[engine_1, engine_2, engine_3 ... engine_n],
...
)
# Engines execution will be in following order
# [engine_1, engine_2, engine_3 ... engine_n]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...,
...
)
await agent.add(engine_1, engine_2, execute_type='SEQUENCE')
.
.
.
await agent.add(engine_3, engine_4, execute_type='SEQUENCE')
.
.
# Engines execution will be in following order
# [engine_1, engine_2, engine_3, engine_4]
```
### Parallel
In parallel execution, `engine(s)` run at the same time, speeding up the process by utilizing multiple threads or processes.
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...
engines=[[engine_1, engine_2, engine_3 ... engine_n]],
...
)
# Engines execution will be in following order
# [[engine_1, engine_2, engine_3 ... engine_n]]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...,
...
)
await agent.add(engine_1, engine_2, execute_type='PARALLEL')
.
.
await agent.add(engine_3, engine_4, execute_type='PARALLEL')
.
.
# Engines execution will be in following order
# [[agent_1, engine_2, engine_3, engine_4]]
```
### Mixed with Sequence and Parallel
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...
engines=[engine_1, [engine_2, engine_3], engine_4 ... engine_n],
...
)
# Engines execution will be in following order
# [engine_1, [engine_2, engine_3], engine_4, ... engine_nm]]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agent import agent
agent = Agent(
...,
...
)
await agent.add(engine_1, engine_2, execute_type='SEQUENCE')
.
.
.
await agent.add(engine_3, engine_4, execute_type='PARALLEL')
.
.
# Engines execution will be in following order
# [engine_1, engine_2, [engine_3, engine_4]]
```
## Task Agent
In SuperAgentX, a task agent is a focused, self-contained component that executes a complete workflow from start to
finish. It receives an input or trigger, applies decision logic, and uses SuperAgentX handlers and tools to interact
with external systems such as APIs, databases, or messaging services. Task agents are typically asynchronous and
event-driven, enabling reliable automation of monitoring, processing, and orchestration tasks with minimal human
intervention.
```python theme={null}
task_agent = Agent(engines=[engine])
```
## Browser Agent
In SuperAgentX, a browser agent is a specialized agent designed to interact with web interfaces just like a human user.
It can navigate pages, fill forms, click buttons, extract data, and handle multi-step web workflows using a controlled
browser environment. Browser agents are used when APIs are unavailable or insufficient, enabling automated web actions
such as scraping, verification, onboarding, and portal-based operations within a larger task workflow.
```python theme={null}
browser_agent = Agent(
name="Review & Pay Agent",
role="Browser Payment Executor",
goal="Review checkout and pay using credit card",
llm=llm,
human_approval=True, # governance point
prompt_template=browser_prompt,
engines=[browser_engine],
max_retry=2
)
```
# Conversation
Source: https://docs.superagentx.ai/conversation
# ๐ง Conversation with Memory
A **conversation with memory** is more than just talkโit's connection.
This guide demonstrates how to create an AI conversation pipeline using the `superagentx` framework, with built-in support for **memory** and **conversation context** through a unique `conversation_id`.
The core idea is to enable an AI agent to **remember the flow of a conversation** using a memory component tied to a specific session.
## 1. Define the AI Handler Pipeline
The function `get_test_ai_handler_pipe` initializes the AI pipeline with:
* An LLM client (using OpenAI)
* A memory module to retain context
* A custom handler to manage LLM interactions
* An agent defined with a goal and role
* A pipeline (`AgentXPipe`) for integration into CLI/WebSocket/API interfaces
```python theme={null}
import uuid
from superagentx.agent import Agent
from superagentx.agentxpipe import AgentXPipe
from superagentx.engine import Engine
from superagentx.llm import LLMClient
from superagentx.memory import Memory
from superagentx.prompt import PromptTemplate
from df_agentic_app.handlers.ai import AIHandler
async def get_test_ai_handler_pipe() -> AgentXPipe:
# LLM Configuration
llm_config = {
'llm_type': 'openai'
}
llm_client = LLMClient(llm_config=llm_config)
# Enable Memory
memory = Memory(memory_config={"llm_client": llm_client})
ai_handler = AIHandler(
llm=llm_client,
memory=memory
)
# Prompt Template
prompt_template = PromptTemplate()
# Create Engine
ai_engine = Engine(
handler=ai_handler,
llm=llm_client,
prompt_template=prompt_template
)
# Create Agents
ai_agent = Agent(
name='AI agent',
goal="Get me the best results",
role="You are the best at answering",
llm=llm_client,
prompt_template=prompt_template,
engines=[ai_engine]
)
# Expose via pipe for interaction
pipe = AgentXPipe(
agents=[ai_agent],
memory=memory
)
return pipe
```
***
## 2. Run a Conversation with Context
This example initializes a pipeline and runs a conversation with memory:
```python theme={null}
import asyncio
import uuid
conversation_id = uuid.uuid4().hex # Unique ID for tracking this conversation
async def main():
conversation_pipe = await get_test_ai_handler_pipe()
await conversation_pipe.flow(
query_instruction="What is Agentic AI?",
conversation_id=conversation_id # Context maintained here
)
await conversation_pipe.flow(
query_instruction="Give me real-world examples of Agentic AI applications?",
conversation_id=conversation_id
)
await conversation_pipe.flow(
query_instruction="How does memory improve AI conversations?",
conversation_id=conversation_id
)
```
if **name** == "**main**":
asyncio.run(main\_conversations())
````
---
โ
**Supports contextual memory** per `conversation_id`
โ
**Ideal for multi-turn chat experiences** across CLI, WebSocket, or RESTful APIs
---
## ๐ช Sample Output
```shell
Generate 'the response based on the user input. When you response the answer,
remember the previous context if previous context is not empty otherwise response it'.
User Input: What is Agentic AI?
Previous Context: []
name='AI agent',
agent_id='d03230eff9084e0097141d24289bbd2c',
reason='The provided Output_Context explains what Agentic AI is, describing its agency,
autonomy, decision-making capabilities, and ability to learn and adapt. This aligns with
the goal of getting the best results by supplying a comprehensive understanding of Agentic AI.',
result='Agentic AI refers to artificial intelligence systems that possess a form of agency,
which means they are designed to act autonomously on behalf of users or stakeholders in an
environment. This kind of AI is capable of decision-making, planning, and executing tasks
without direct human intervention, often using advanced algorithms to optimize for specific
outcomes. Unlike purely reactive or rule-based systems, agentic AI can adapt to new situations,
learn from interactions, and make complex decisions by considering various factors and potential
consequences.',
content=None,
error=None,
is_goal_satisfied=True
User Input: Give me real-world examples of Agentic AI applications?
Previous Context: []
name='AI agent',
agent_id='177fdd1a9704457b981bed5db33858fc',
reason='The provided Output_Context contains a clear explanation of Agentic AI, listing specific real-world applications that demonstrate AI systems
possessing agency. These examples comprehensively illustrate the autonomy and decision-making capabilities inherent in Agentic AI.',
result=[
1.Autonomous Vehicles: Such as self-driving cars that navigate and make driving decisions without human intervention.
2.Robotic Process Automation (RPA): Advanced systems that make autonomous decisions within business processes.
3.Drones and Unmanned Aerial Vehicles (UAVs): Used for delivery, surveillance, or search and rescue operations.
4.Smart Personal Assistants: Like Google Assistant, Siri, or Alexa, which make proactive suggestions or actions.
5.Intelligent Customer Support Bots: Bots that manage complex inquiries and escalate issues autonomously.
6.Industrial Robots: Robots in manufacturing that adapt and optimize tasks autonomously.
7.AI in Financial Trading Systems: Systems making autonomous stock trading decisions.
],
content=None,
error=None,
verify_goal=True,
is_goal_satisfied=True
User Input: How does memory improve AI conversations?
Previous Context: []
name='AI agent',
agent_id='177fdd1a9704457b981bed5db33858fc',
reason='The Output_Context does not address the query about how memory improves AI conversations. Instead, it discusses Agentic AI and its applications,
which are not relevant to the query. The additional provided response outside the context clearly addresses how memory can enhance AI conversations by explaining
various aspects like context retention, personalization, learning from interactions, reducing repetition, and improved decision-making.
result="Memory can enhance AI conversations in several ways:\n\n1. **Context Retention**: Memory allows an AI to retain the context of an ongoing
conversation, enabling more coherent and contextually relevant responses. By remembering earlier interactions, an AI can more effectively follow a thread of
conversation and reference past exchanges to enhance the dialogue.\n\n2. **Personalization**: With memory, an AI can remember users' preferences, personal
details, previous questions, and feedback. This leads to personalized interactions, as the AI can tailor its responses based on individual user profiles or
specific user history.\n\n3. **Learning from Interactions**: Memory enables an AI to learn from past interactions and improve over time. As it stores information
about what worked well in previous conversations or where misunderstandings occurred, it can adapt its approach to become more effective in future
interactions.\n\n4. **Reducing Repetition**: By recalling earlier interactions, an AI can avoid repetitive questioning or information exchange, which improves
user experience. Users do not have to repeat themselves if the AI remembers what has already been discussed.\n\n5. **Improved Decision Making**: Memory allows an
AI to remember previous successes or failures, which can inform its decision-making processes. By leveraging past experiences, the AI can choose strategies that
are more likely to be effective in achieving desired outcomes.\n\nOverall, memory enhances the quality and effectiveness of AI conversations by adding continuity,
depth, and adaptability to interactions.",
content=None,
error=None,
verify_goal=True,
is_goal_satisfied=True
````
***
## ๐ง Why Memory Matters
Memory enables **multi-turn conversations** where the AI doesn't forget previous questions, responses, or the conversation's purpose.
This is essential for building helpful and coherent AI assistants that feel **more natural** and **less robotic**.
# Engines
Source: https://docs.superagentx.ai/engines
An `Engine` is a system that sets up and runs specific processes to perform `handler`. It organizes different `handler`
functions and uses a given prompt along with a language model to activate them. Essentially, the `engine` helps
manage and trigger these processes to carry out actions based on the instructions it receives.
### Parameters
| Attribute | Parameters | Description |
| :----------------------------- | :---------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Handler** | `handler` | Implementation of `BaseHandler`, its method(s) will be executed based on the given prompts. |
| **LLM Client** | `llm` | Interface for communicating with the large language model (LLM). |
| **Prompt Template** | `prompt_template` | Defines the structure and format of prompts sent to the LLM using `PromptTemplate`. |
| **Tools** *(optional)* | `tools` | List of handler method names (as dictionaries or strings) available for use during interactions. Defaults to `None`. If nothing provide `Engine` will get it dynamically using `dir(handler)`. |
| **Output Parser** *(optional)* | `output_parser` | An optional parser to format and process the handler tools output. Defaults to `None`. |
```python theme={null}
from superagentx.engine import Engine
walmart_engine = Engine(
handler=walmart_ecom_handler,
llm=llm_client,
prompt_template=prompt_template,
tools=None,
output_parser=None,
)
```
## Browser Engine
The `BrowserEngin`e is a specialized engine designed to execute browser-based automation tasks using a set of
predefined browser handlers. It coordinates a language model, a prompt template, and a browser context to perform
interactive web actions such as navigation, input, clicking, scrolling, and data extraction. By interpreting
instructions and invoking the appropriate browser tools, the `BrowserEngine` enables intelligent and automated
web browsing workflows, including optional screenshot capturing and step-based execution control.
### Parameters
| Attribute | Parameters | Description |
| :------------------------------------- | :---------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **LLM Client** | `llm` | An instance of LLMClient, which likely represents a client or interface to interact with a Large Language Model (LLM) such as GPT. This is used for generating text, answering queries, etc. |
| **PromptTemplate** | `prompt_template` | An instance of `PromptTemplate`, which defines the format or structure of the prompts that will be sent to the LLM. It helps maintain consistency and clarity in how requests are made to the model. |
| **Browser Instance Path** *(optional)* | `browser_instance_path` | If we want to reuse a browser session that was saved earlier, this tells the program where to find it. Defaults to `None` |
| **Browser** *(optional)* | `browser` | A web browser (like Chrome or Firefox) that the program can control automatically. We can use this if we need to interact with websites without doing it manually. Defaults to `None` |
| **Browser Context** *(optional)* | `browser_context` | A separate "space" inside the browser that keeps things organized. Itโs like using different tabs in the browser for different tasks or projects. Defaults to `None` |
| **Tools** *(optional)* | `tools` | A list of special tools (like a calculator, search tool, or others) that the program can use during its tasks. Defaults to `None` |
| **Max Steps** | `max_steps` | The maximum number of steps the program can take before it stops. This helps prevent it from running forever or getting stuck in endless loops. Default Step is `100` |
| **Screen Shot** *(optional)* | `take_screenshot` | A yes/no option that tells the program whether or not to take screenshots during its work. This is useful for keeping a visual record of what happens.Defaults to `None` |
| **Screenshot Path** *(optional)* | `screenshot_path` | If screenshots are taken, this tells the program where to save them on your computer. Defaults to `None` |
```python BrowserEngine: theme={null}
from superagentx.browser_engine import BrowserEngine
browser_engine = BrowserEngine(
llm=llm_client,
prompt_template=prompt_template,
)
```
### Browser Configuration:
`BrowserConfig` controls the global launch behavior and core settings for the browser instance.
It specifies parameters like headless mode, browser type, proxy settings, and remote connections.
This configuration is applied when the browser starts and affects all created contexts and pages.
```python Browser config theme={null}
from superagentx.computer_use.browser.browser import Browser, BrowserConfig
config = BrowserConfig(
headless=False,
disable_security=True,
browser_type='firefox'
)
browser = Browser(config=config)
```
#### BrowserConfig Parameters:
| Attribute | Parameters | Description |
| :--------------------------------------- | :------------------------------ | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Headless** | `headless` | When True, launches the browser without a visible UI. Useful for automation or CI environments, but some sites may detect headless mode. Default to `False`. |
| **Disables Security** | `disable_security` | Disables standard browser security features (e.g., CORS, mixed content blocking). Use only when necessary and only with trusted content. Default to `True`. |
| **extra\_chromium\_args** *(optional)* | `extra_chromium_args` | Additional command-line flags to pass to Chromium during launch (e.g., --no-sandbox, --disable-gpu).Default is \[]. |
| **chrome\_instance\_path** *(optional)* | `chrome_instance_path` | Path to a custom Chrome or Chromium executable. Useful when using a non-default installation. Default to `None`. |
| **wss\_url** *(optional)* | `wss_url` | WebSocket endpoint for connecting to a remote browser instance via the DevTools Protocol (e.g., for cloud or containerized browsers). Default to `None`. |
| **cdp\_url** *(optional)* | `cdp_url` | Chrome DevTools Protocol (CDP) endpoint used for remote browser automation. Default to `None`. |
| **Proxy** *(optional)* | `proxy` | Configuration for routing browser traffic through a proxy server. Follows Playwright-style proxy settings (e.g., server, username, password). Default to `None`. |
| **New Content Config** *(optional)* | `new_context_config` | Default configuration for new browser contexts. Controls cookies, viewport, locale, user agent, and other context-level behaviors. Default is `BrowserContextConfig()`. |
| **\_force\_keep\_browser\_alive** | \`\_force\_keep\_browser\_alive | Internal flag to keep the browser instance running across multiple tasks or scripts. Default to `False`. |
| **Browse Type** | `browser_type` | Specifies which browser engine to use: 'chromium', 'firefox', or 'webkit'. In this example, 'firefox' is selected. Default is `chromium`. |
#### Browser Context Configuration:
`BrowserContextConfig` defines the default environment and behavior for each new browser context.
It includes settings like cookies, viewport size, user agent, locale, and download paths.
This configuration ensures isolated, customizable sessions within a single browser instance.
```python theme={null}
from superagentx.computer_use.browser.context import BrowserContextConfig
config = BrowserContextConfig(
cookies_file="path/to/cookies.json",
wait_for_network_idle_page_load_time=3.0,
window_width=1280,
window_height=1100,
locale='en-US',
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.102 Safari/537.36',
highlight_elements=True,
viewport_expansion=500,
allowed_domains=['google.com', 'wikipedia.org'],
)
browser = Browser()
context = BrowserContext(browser=browser, config=config)
browser_engine = BrowserEngine(
llm=llm_client,
prompt_template=prompt_template,
browser_context=context
)
```
#### BrowserContextConfig Parameters:
| Attribute | Parameters | Description |
| :--------------------------------------- | :------------------------------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Cookies File** *(optional)* | `cookies_file` | Path to a file containing cookies to be loaded into the browser session. Useful for persisting login states. Default to `None`. |
| **Minimum Wait Page Load Time** | `minimum_wait_page_load_time` | Minimum time (in seconds) to wait after page navigation begins before proceeding. Ensures smoother page loading. Default to `0.25` |
| **Wait For Network Idle Page Load Time** | `wait_for_network_idle_page_load_time` | Additional wait time after the network becomes idle, simulating user behavior more closely. Default to `0.5` |
| **Maximum Wait Page Load Time** | `maximum_wait_page_load_time` | Maximum duration to wait for a page to fully load. Acts as a timeout for slow-loading pages. Default to `5`. |
| **Wait Between Actions** | `wait_between_actions` | Wait time between sequential browser actions (e.g., clicks, navigation). Helps avoid detection by anti-bot systems. Default to `0.5` |
| **Disable Security** *(optional)* | `disable_security` | Disables browser-level security features. Can solve cross-origin issues, but should not be used with untrusted websites. Default to `True` |
| **Browser Window Size** *(optional)* | `browser_window_size` | Sets the browser window size. Useful for rendering consistency and testing responsive layouts. Default to `{'width': 1280, 'height': 1100}` |
| **No Viewport** *(optional)* | `no_viewport` | When set to True, disables Playwrightโs default viewport setting and uses the system window size instead. Default to `None`. |
| **Save Recording Path** *(optional)* | `save_recording_path` | Path to save screen recordings of the browser session, if enabled. Default to `None`. |
| **Save Downloads Path** *(optional)* | `save_downloads_path` | Directory path to store downloaded files from within the browser context. Default to `None`. |
| **Trace Path** *(optional)* | `trace_path` | Path to store browser traces (used for debugging and performance analysis). Default to `None`. |
| **Locale** *(optional)* | `locale` | Sets the locale of the browser context (e.g., "en-US"), which can affect how websites render language and date formats.Default to `None`. |
| **User Agent** | `user_agent` | The user agent string used for the browser session. Can be customized to mimic specific devices or browsers. Default to `realistic Chrome UA string`. |
| **Highlight Elements** | `highlight_elements` | Highlights elements being interacted with in the browser. Useful for debugging and visual confirmation during automation. Default to `False` |
| **Viewport Expansion** | `viewport_expansion` | Expands the viewport vertically (in pixels) to ensure that off-screen elements are brought into view. Default is 500. |
| **Allowed Domains** *(optional)* | `allowed_domains` | List of domains that are permitted to load. Helps limit unwanted navigation or third-party content. Default to `None`. |
| **Include Dynamic Attributes** | `include_dynamic_attributes` | Whether to include dynamically generated attributes (like data-*, aria-*) during automation tasks. Default to `True`. |
| **Force Keep Context Alive** | `_force_keep_context_alive` | Internal setting to keep the browser context alive beyond the usual lifecycle. Useful for chained or long-lived tasks. Default to `False`. |
# Introduction
Source: https://docs.superagentx.ai/examples-reference/introduction
A set of examples that demonstrate how to use the SuperAgentX framework to automate your workflows.
Ecommerce is the buying and selling of goods or services over the internet.
Retrieving real-time or forecasted weather information from external sources.
Find nearby cafes with detailed information about amenities, opening hours, and features.
To automate trip planning when you're deciding between multiple options.
This user-friendly smart app helps outpatients identify the best-specialized doctor and book an appointment with the doctor simultaneously.
# Create Handler
Source: https://docs.superagentx.ai/examples/create_handler
Here's a step-by-step guide to creating a new custom handler in `SuperAgentX`, inheriting from `BaseHandler` (`superagentx.handler.base.BaseHandler`), to enhance and extend the system's functionality.
### Step 1: Creating a New Handler
Now, we will create a new class `FileHandler` that inherits from `BaseHandler`. This class will handle reading a text file
and filtering the contents based on a search query.
Here's the FileHandler implementation:
```python theme={null}
import asyncio
import aiofiles
from superagentx.handler.base import BaseHandler
from superagentx.handler.decorators import tool
class FileHandler(BaseHandler):
def __init__(self, file_path: str):
"""
Initializes the FileHandler with the given file path.
Args:
file_path (str): The path to the text file to be read.
"""
super().__init__() # Call the parent class's constructor
self.file_path = file_path # Store the file path for later use
```
### Step 2: Define Handler Method(s).
Now, letโs define a method inside the handler to perform an action, like processing data from the given method.
This `tool` (`superagentx.handler.decorators.tool`) decorator is used to preserve llm tool function
when you wrap it in a new function, often used in custom decorators.
Write a docstring for the method based on the parameters provided. The docstring should be clear, detailed,
and understandable by an LLM (Large Language Model). The description should be 3 to 4 lines long with a brief
explanation, and parameter can be in a single line.
```python theme={null}
@tool
async def read_file(self):
"""
Reads the contents of the file specified by the file path.
Returns:
list: A list of lines from the file.
"""
try:
async with aiofiles.open(self.file_path, 'r') as file:
lines = await file.readlines()
return lines
except FileNotFoundError:
print(f"Error: The file at {self.file_path} was not found.")
return []
@tool
async def search(
self,
query: str
):
"""
Searches for lines in the file that contain the provided query.
Args:
query (str): The string to search for in the file.
Returns:
list: A list of lines containing the query.
"""
lines = await self.read_file() # Get lines from the file
filtered_lines = [line for line in lines if query.lower() in line.lower()] # Filter lines by query
return filtered_lines
```
### Step 3: Testing the Handler
To test the FileHandler class, create a sample text file and use the handler to search for specific terms.
## create the sample.txt file
Create a file named `sample.txt` in the same directory as your script with the following content:
```text theme={null}
Hello, welcome to the file handler example.
This file contains several lines of text.
You can search through it using queries.
The FileHandler will help you find matching lines.
```
### Use the `FileHandler` class to search for the term `"file"`:
```python theme={null}
# Create an instance of FileHandler with the path to your text file
file_handler = FileHandler(file_path='sample.txt')
# Search for the term 'file' in the file
search_result = file_handler.search('file')
# Print the search results
print("Search Results:")
for line in search_result:
print(line.strip()) # Strip removes the newline character
```
### Full Code
```python theme={null}
import asyncio
import aiofiles
from superagentx.handler.base import BaseHandler
from superagentx.handler.decorators import tool
class FileHandler(BaseHandler):
def __init__(self, file_path: str):
"""
Initializes the FileHandler with the given file path.
Args:
file_path (str): The path to the text file to be read.
"""
super().__init__() # Call the parent class's constructor
self.file_path = file_path # Store the file path for later use
@tool
async def read_file(self):
"""
Reads the contents of the file specified by the file path.
Returns:
list: A list of lines from the file.
"""
try:
async with aiofiles.open(self.file_path, 'r') as file:
lines = await file.readlines()
return lines
except FileNotFoundError:
print(f"Error: The file at {self.file_path} was not found.")
return []
@tool
async def search(
self,
query: str
):
"""
Searches for lines in the file that contain the provided query.
Args:
query (str): The string to search for in the file.
Returns:
list: A list of lines containing the query.
"""
lines = await self.read_file() # Get lines from the file
filtered_lines = [line for line in lines if query.lower() in line.lower()] # Filter lines by query
return filtered_lines
# Create an instance of FileHandler with the path to your text file
file_handler = FileHandler(file_path='sample.txt')
if __name__ == "__main__":
# Search for the term 'file' in the file
search_result = asyncio.run(file_handler.search('file'))
# Print the search results
print("Search Results:")
for line in search_result:
print(line.strip()) # Strip removes the newline character
```
### Expected Output:
When you run this code, it will output:
```text theme={null}
Search Results:
This file contains several lines of text.
You can search through it using queries.
The FileHandler will help you find matching lines.
```
# Amazon Handler
Source: https://docs.superagentx.ai/examples/handlers/E-commerce/amazon
AmazonHandler, for interacting with the Amazon data API using RapidAPI. It uses to fetch Amazon product
data and customer reviews.
## Example
This code performs a product search on Amazon. It uses the AmazonHandler to connect to Amazonโs system and returns
relevant product results and reviews based on the provided query.
To access RapidAPI services, you'll need to create a RapidAPI account, obtain an API key, and subscribe the required
API for your application. For more details refer here.
> Set up a RapidAPI key as an environmental variable and run the following code.
```python theme={null}
export RAPID_API_KEY = "**************************"
```
```python amazon_handler.py theme={null}
import asyncio
import os
from superagentx_handlers.ecommerce.amazon import AmazonHandler
async def search(query):
amazon = AmazonHandler()
return await amazon.product_search(query=query)
query = input("Enter your query here: ")
print(asyncio.run(search(query)))
```
## Result
```inline amazon theme={null}
Enter Query: give a best juice blender under 2k
[ { 'asin': 'B0DHXZQDD2',
'climate_pledge_friendly': False,
'currency': 'INR',
'delivery': 'FREE delivery Tue, 22 Oct Or fastest delivery Mon, 21 Oct',
'has_variations': False,
'is_amazon_choice': False,
'is_best_seller': False,
'is_prime': True,
'product_minimum_offer_price': 'โน659',
'product_num_offers': 1,
'product_num_ratings': 31,
'product_original_price': 'โน1,599',
'product_photo': 'https://m.media-amazon.com/images/I/714Xz3C8tNL._AC_UY654_FMwebp_QL65_.jpg',
'product_price': 'โน659',
'product_star_rating': '5',
'product_title': 'ROMINO 380ml USB Rechargeable Mini Juicer Blender, '
'Portable Juicer Bottle, Electric Fruit Juice Maker '
'Machine, Personal Size Juicer Bottle Blender Grinder '
'Mixer for Juices, Shakes & Smoothies (Multi)',
'product_url': 'https://www.amazon.in/dp/B0DHXZQDD2',
'reviews': [],
'sales_volume': '400+ bought in past month'},
{ 'asin': 'B0DFWCM981',
'climate_pledge_friendly': False,
'currency': 'INR',
'delivery': 'FREE delivery Sun, 20 Oct Or fastest delivery Tomorrow, '
'19 Oct',
'has_variations': False,
'is_amazon_choice': False,
'is_best_seller': False,
'is_prime': True,
'product_minimum_offer_price': 'โน999',
'product_num_offers': 1,
'product_num_ratings': 63,
'product_original_price': 'โน2,599',
'product_photo': 'https://m.media-amazon.com/images/I/61AcxigXJvL._AC_UY654_FMwebp_QL65_.jpg',
'product_price': 'โน999',
'product_star_rating': '4.4',
'product_title': 'SUPERSTUD Portable USB Juicer Blender for Juices and '
'Smoothie, Milk Shakes, 380ML, built-in Jar (Rich '
'Grey)',
'product_url': 'https://www.amazon.in/dp/B0DFWCM981',
'reviews': [],
'sales_volume': 'M.R.P: '}]
```
# Best Buy Handler
Source: https://docs.superagentx.ai/examples/handlers/E-commerce/best_buy
Best Buy is a leading consumer electronics retailer that provides a wide variety of products including laptops, TVs, phones, appliances, and more.
The **BestbuyHandler** allows programmatic access to Best Buyโs product catalog through its public API. Developers can search products, retrieve prices, discounts, reviews, and detailed product information.
This integration is useful for building product recommendation systems, price tracking tools, or e-commerce integrations.
## Example
To create the `BestbuyHandler` object with the API key:
```python theme={null}
import os
from superagentx_handlers.retail.bestbuy import BestbuyHandler
bestbuy_handler = BestbuyHandler(
api_key=os.getenv("BESTBUY_API_KEY")
)
```
**Search Products (get\_best\_buy\_info):**
Fetch product details from Best Buy based on a search keyword.
```python theme={null}
products = await bestbuy_handler.get_best_buy_info(search_text="laptop")
print(products)
```
# Flipkart Handler
Source: https://docs.superagentx.ai/examples/handlers/E-commerce/flipkart
Flipkart, a leading e-commerce platform in India, offers millions of products across categories
like electronics, fashion, home appliances, and groceries. Efficient product search is crucial
to its ecosystem, helping users quickly discover relevant items through filters, recommendations,
and keyword-based search functionality.
## Flipkart Code
This code searches for products on Flipkart. It uses the FlipkartHandler to connect to Flipkartโs platform and returns
product results based on the given query.
To access RapidAPI services, you'll need to create a RapidAPI account, obtain an API key, and subscribe the required
API for your application. For more details refer here.
> Set up a RapidAPI key as an environmental variable and run the following code.
```python theme={null}
export RAPID_API_KEY = "**************************"
```
```python flipkart_handler.py theme={null}
import asyncio
import os
from superagentx_handlers.ecommerce.flipkart import FlipkartHandler
async def search(query):
flipkart = FlipkartHandler()
return await flipkart.product_search(query=query)
query = input("Enter your query here: ")
asyncio.run(search(query))
```
## Result:
```inline flipkart theme={null}
Enter Query: oneplus
[{'pid': 'ACCGXAV7Z868V64N', 'itemId': 'ITM617E66D574EA4', 'listingId': 'LSTACCGXAV7Z868V64NMNDSVH',
'brand': 'Flipkart SmartBuy', 'title': 'Flipkart SmartBuy USB Type C Cable 8 A 1 m Silver Wrap OEM 65W-10W/6.
5A VOOC/WARP/DASH/SUPERVOOC/SUPERDART CHARGER', 'subTitle': 'Compatible with oppo,realme,narzo,oneplus,vivo,iqoo,
samsung,motorola,mi,redmi,poco, Red, White, One Cable', 'url': 'https://flipkart.com/flipkart-smartbuy-usb-type-c-
cable-8-1-m-silver-wrap-oem-65w-10w-6-5a-vooc-warp-dash-supervooc-superdart-charger/p/itm617e66d574ea4?
pid=ACCGXAV7Z868V64N', 'badge': None, 'stock': 'IN_STOCK', 'highlights': ['Length 1 m', 'Round Cable',
'Connector One: USB Type A|Connector Two: Type C', 'Cable Speed: 980 Mbps', 'Mobile, Laptop, Power Bank, Wireless
Earphones, Speaker, Trimmer, Tablet'], 'isSponsored': False, 'mrp': 999, 'price': 199, 'rating': {'average': 4.1,
'count': 15168, 'reviewCount': 684, 'breakup': [1408, 596, 1330, 3449, 8385]}, 'images': ['https://rukminim2.
flixcart.com/image/1160/1160/xif0q/data-cable/usb-type-c-cable/9/c/3/oem-65w-10w-6-5a-vooc-warp-dash-supervooc-
superdart-charger-original-imah52f2rw3f6gzz.jpeg?q=90', 'https://rukminim2.flixcart.com/image/1160/1160/xif0q/data-
cable/usb-type-c-cable/f/c/p/oem-65w-10w-6-5a-vooc-warp-dash-supervooc-superdart-charger-original-imahfa9cugzrfhya.
jpeg?q=90', 'https://rukminim2.flixcart.com/image/1160/1160/xif0q/data-cable/usb-type-c-cable/4/z/v/oem-65w-10w-6-5a-
vooc-warp-dash-supervooc-superdart-charger-original-imahfa9cwmgq7kwq.jpeg?q=90', 'https://rukminim2.
flixcart.com/image/1160/1160/xif0q/data-cable/usb-type-c-cable/c/z/x/oem-65w-10w-6-5a-vooc-warp-dash-supervooc-
superdart-charger-original-imahfa9cbgtqswwa.jpeg?q=90'], 'reviews': [{'rating': 4, 'title': 'Nice product',
'review': 'I am using Realme gt ME.This cable is working for 65w Charger', 'images': [], 'reviewer': 'sarath kumar',
'location': 'Hosur, Tamil Nadu', 'upvotes': 57, 'downvotes': 10, 'date': '6 months ago'}, {'rating': 5, 'title':
'Simply awesome', 'review': 'Support 67 watt charging. Like it', 'images': [], 'reviewer': 'Satendra Kumar',
'location': 'Baghpat, Uttar Pradesh', 'upvotes': 28, 'downvotes': 6, 'date': '6 months ago'}
```
# Walmart Handler
Source: https://docs.superagentx.ai/examples/handlers/E-commerce/walmart
A Walmart handler is a tool that makes it easy for developers to connect their apps or websites to Walmartโs system.
With this tool, businesses can access Walmartโs product listings, prices, stock levels, and availability in real-time.
It helps automate tasks like searching for products, tracking inventory, and monitoring price changes.
This handler is very useful for e-commerce websites, price comparison tools, and inventory management systems.
## Example
This code sets up a WalmartHandler, which allows interacting with Walmartโs platform to search for products or access
related data programmatically.
To access RapidAPI services, you'll need to create a RapidAPI account, obtain an API key, and subscribe the required
API for your application. For more details refer here.
> Set up a RapidAPI key as an environmental variable and run the following code.
```python theme={null}
export RAPID_API_KEY = "**************************"
```
```python walmart_handler.py theme={null}
import os
import asyncio
from superagentx_handlers.ecommerce.walmart import WalmartHandler
async def search(query: str):
walmart_handler = WalmartHandler()
return await walmart_handler.product_search(query=query)
query = input("Enter your query here: ")
asyncio.run(search(query))
```
## Result
```python walmart theme={null}
{'body': {'pagination': {'currentPage': 1, 'nextPage': 2, 'totalPages': 2},
'products': [{'image': 'https://i5.walmartimages.com/seo/VAVSEA-1000W-Countertop-Blender-for-Shake-and-Smoothies-for-Ice-Crush-Fruit-Protein-Drinks-Vegetable_7357b6c5-180c-4724-a87f-6c3aa4478ea7.55c0e5952798f4cb3fa4f542e91ccce9.jpeg?odnHeight=784&odnWidth=580&odnBg=FFFFFF',
'isBestSeller': False,
'link': 'https://www.walmart.com/sp/track?bt=1&eventST=click&plmt=sp-search-middle~desktop~&pos=1&tax=4044_90548&rdf=1&rd=https%3A%2F%2Fwww.walmart.com%2Fip%2FVAVSEA-1000W-Countertop-Blender-for-Shake-and-Smoothies-for-Ice-Crush-Fruit-Protein-Drinks-Vegetable%2F5508034320%3FclassType%3DVARIANT%26adsRedirect%3Dtrue&adUid=783f604f-db6b-447c-868e-f2754d136c63&mloc=sp-search-middle&pltfm=desktop&pgId=best%20one%20blender%20provide%205%20ratings&pt=search&spQs=Bh8fdeN9dkO0SpwJt6bexW-EJPUbTycqL1ZKTTwarC0NEfhAGg0Fbpr0X-_cGvl5jvxi1R7_GdjNEpdJf2fSWVdx3hbA2qIUSzwtQ06Pfj3fQCLeitKYJzKbWWKEgyULBf0-crnzapKUG9-C85yCe0y37G3LOnPI6QWXRrLIiIyBiBBB1749_CYKqswwKpZXrTHmwuqTRSxBi7l3KJXHtcQ0UJl7hiNx8z0jcfxHQeRWi7f_DU7vNw2RhH9uWOVjkneYhOD_GK-qwlh4Wmw0kg&storeId=2038&couponState=na&bkt=ace1_11168%7Cace2_11182%7Cace3_11181%7Ccoldstart_off%7Csearch_default&classType=VARIANT',
'outOfStock': False,
'position': 1,
'price': {'currency': '$',
'currentPrice': '$56.99',
'originalPrice': '$179.99',
'rawPrice': '56.99'},
'ratings': '4.8',
'reviewsCount': '60',
'shippingMessage': 'Free shipping, arrives in 2 days',
'title': 'VAVSEA 1200W Blender for Smoothies and '
'Shakes, Personal Kitchen Blenders Mixer '
'Grinder with 2 Travel Cups 1 Grinding Cup '
'for Ice Crush Frozen Drinks, BPA-Free'}],
'productsCount': 49,
'resultInfo': 'Results for "best one blender provide 5 ratings"(60)'},
'original_status': 200,
'pc_status': 200,
'url': 'https://www.walmart.com/search?q=best one blender provide 5 ratings'}
```
# API Gateway Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/api_gateway
Google Cloud API Gateway is a fully managed service that allows developers to create, secure, and monitor APIs for their applications. It enables seamless management of APIs at scale, integrates with IAM for authentication, and provides features like quota management, monitoring, and logging.
The `GCPAPIGatewayHandler` helps interact with Google Cloudโs API Gateway service to fetch gateway details programmatically.
## Example
To create the `GCPAPIGatewayHandler` object, initialize it with your Google service account credentials (JSON file path or dictionary).
```python theme={null}
from superagentx_handlers.gcp.api_gateway import GCPAPIGatewayHandler
gateway_handler = GCPAPIGatewayHandler(
creds="service-account.json" # or dict with service account info
)
```
**List API Gateways:**
Lists all API Gateways available in your GCP project across supported regions.
```python theme={null}
gateways = await gateway_handler.list_gateways(page_size=50)
for g in gateways:
print(g["name"], g["state"], g["default_hostname"])
```
**Convert Gateway to Dictionary (\_gateway\_to\_dict):**
Converts a Gateway object into a structured dictionary with useful metadata.
```python theme={null}
gateway_obj = (await gateway_handler.client.list_gateways(
request={"parent": "projects/my-project/locations/us-central1"}
)).pages[0][0]
gateway_dict = await gateway_handler._gateway_to_dict(gateway_obj, location="us-central1")
print(gateway_dict)
```
**Get Gateway State (\_get\_gateway\_state):**
Converts a raw state value (string, int, or enum) into a Gateway.State enum.
```python theme={null}
from google.cloud import apigateway_v1
# Using string
print(_get_gateway_state("CREATING").name) # Output: CREATING
# Using integer (matches enum value)
print(_get_gateway_state(1).name) # Output: ACTIVE
# Using enum directly
print(_get_gateway_state(apigateway_v1.Gateway.State.FAILED).name) # Output: FAILED
```
# Calendar Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/calendar
Google Calendar is a scheduling and time-management service that allows users to create, manage, and share events. It provides features like reminders, recurring events, and integration with other Google Workspace services. With the `CalenderHandler`, you can connect to Google Calendar via OAuth 2.0 and programmatically fetch events for a day, week, month, or specific type.
## Example
To create the `CalenderHandler` object, initialize it with your Google OAuth credentials file. The handler manages authentication and provides methods to retrieve events from Google Calendar.
```python theme={null}
import os
from superagentx_handlers.gcp.calendar import CalenderHandler
calendar_handler = CalenderHandler(
credentials="credentials.json" # Path to Google OAuth client secrets file
)
```
**Get Todayโs Events:**
Retrieves all events scheduled for today.
```python theme={null}
events_today = await calendar_handler.get_today_events()
print(events_today)
```
**Get Week Events:**
Retrieves all events scheduled for the next 7 days.
```python theme={null}
events_week = await calendar_handler.get_week_events()
print(events_week)
```
**Get Month Events:**
Retrieves all events scheduled for the next 30 days.
```python theme={null}
events_month = await calendar_handler.get_month_events()
print(events_month)
```
**Get Events by Custom Days:**
Fetches events for a custom range between 1 and 30 days.
```python theme={null}
events_5days = await calendar_handler.get_events_by_days(days=5)
print(events_5days)
```
**Get Events by Type:**
Fetches events of a specific type such as "birthday", "focusTime", or "workingLocation".
```python theme={null}
birthday_events = await calendar_handler.get_events_by_type(event_type="birthday")
print(birthday_events)
```
# Cloud Run Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/cloud_run
Google Cloud Run is a **serverless compute platform** that lets you run containers without managing servers.
It automatically scales up or down based on traffic, integrates with other GCP services, and supports deployments in multiple regions.
Cloud Run abstracts away infrastructure management, allowing you to focus on your application logic.
## Example
To create the `GCPCloudRunHandler`, pass Google service account credentials (either JSON file path or dict).
```python theme={null}
import os
from superagentx.handler.gcp.cloud_run import GCPCloudRunHandler
cloud_run_handler = GCPCloudRunHandler(
creds=os.getenv("GOOGLE_APPLICATION_CREDENTIALS") # Path to service account JSON
)
```
**get\_attr:**
Safely fetches an attribute, dict key, or list index with a default fallback.
```python theme={null}
from superagentx.handler.gcp.cloud_run import get_attr
data = {"name": "cloudrun", "type": "service"}
print(get_attr(data, "name")) # "cloudrun"
print(get_attr(data, "invalid", "N/A")) # "N/A"
```
**get\_nested\_attr:**
Fetches nested attributes safely using a path list.
```python theme={null}
from superagentx.handler.gcp.cloud_run import get_nested_attr
data = {"spec": {"template": {"spec": {"service_account": "default"}}}}
print(get_nested_attr(data, ["spec", "template", "spec", "service_account"]))
# Output: "default"
```
**client (property):**
Lazy initialization of the async Cloud Run client.
```python theme={null}
client = cloud_run_handler.client
print(type(client)) # google.cloud.run_v2.ServicesAsyncClient
```
**close:**
Closes the async client to free resources.
```python theme={null}
await cloud_run_handler.close()
```
**\_process\_container\_info:**
Extracts metadata for a Cloud Run container (image, ports, env vars, resources).
```python theme={null}
container_info = await cloud_run_handler._process_container_info(container_obj)
print(container_info["image"])
```
**\_process\_service\_info:**
Processes a Cloud Run service object into a structured dictionary.
```python theme={null}
service_info = await cloud_run_handler._process_service_info(service_obj, "us-central1")
print(service_info["name"], service_info["url"])
```
**\_fetch\_services\_for\_location:**
Fetches all Cloud Run services in a specific region.
```python theme={null}
services = await cloud_run_handler._fetch_services_for_location("us-central1")
print([s["name"] for s in services])
```
**fetch\_location\_with\_semaphore:**
Helper method to fetch services from a location with concurrency limits.
```python theme={null}
# Example inside async workflow
services = await cloud_run_handler._fetch_services_for_location("us-west1")
```
**get\_gcp\_cloud\_run\_details:**
Lists all Cloud Run services across all supported GCP regions with summary.
```python theme={null}
cloud_run_data = await cloud_run_handler.get_gcp_cloud_run_details()
print(cloud_run_data["summary"])
```
# Compute Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/compute
Google Cloud Compute Engine is an Infrastructure-as-a-Service (IaaS) offering from GCP that allows you to run virtual machines (VMs), manage disks, networks, firewalls, snapshots, and images.
This handler (`GcpComputeHandler`) provides methods to collect detailed information about Compute resources for governance, auditing, and monitoring purposes.
It authenticates using a **GCP service account** and leverages the Google Cloud SDK (`google-cloud-compute`) to fetch data.
## Example
To create the `GcpComputeHandler` object with your service account:
```python theme={null}
import os
import json
from superagentx_handlers.gcp.compute import GcpComputeHandler
# Load service account credentials
with open("service_account.json", "r") as f:
service_account_info = json.load(f)
compute_handler = GcpComputeHandler(service_account_info=service_account_info)
```
**Collect Instances:**
Retrieve details of all VM instances in a project or specific zone.
```python theme={null}
instances = await compute_handler.collect_instances(project_id="my-project-id", zone="us-central1-a")
print(instances[:2]) # preview first 2 VMs
```
**Collect Disks:**
Fetch persistent disk information across all zones or a specific zone.
```python theme={null}
disks = await compute_handler.collect_disks(project_id="my-project-id")
print(disks[:2]) # preview first 2 disks
```
**Collect Networks:**
Get all VPC networks within a project.
```python theme={null}
networks = await compute_handler.collect_networks(project_id="my-project-id")
print(networks)
```
**Collect Subnetworks:**
Retrieve subnetworks across all regions or a specific region.
```python theme={null}
subnets = await compute_handler.collect_subnetworks(project_id="my-project-id", region="us-central1")
print(subnets[:2])
```
**Collect Firewall Rules:**
Fetch firewall rules that define ingress/egress traffic control for resources.
```python theme={null}
firewalls = await compute_handler.collect_firewall_rules(project_id="my-project-id")
print(firewalls[:2])
```
**Collect Images:**
Retrieve custom images created in the project.
```python theme={null}
images = await compute_handler.collect_images(project_id="my-project-id")
print(images)
```
**Collect Snapshots:**
Fetch disk snapshots for backup and disaster recovery.
```python theme={null}
snapshots = await compute_handler.collect_snapshots(project_id="my-project-id")
print(snapshots)
```
**Collect All Compute Info:**
Retrieve a comprehensive dictionary with all compute resources (instances, disks, networks, subnetworks, firewalls, images, and snapshots).
```python theme={null}
all_compute_info = await compute_handler.collect_all_compute_info(project_id="my-project-id")
print(all_compute_info.keys())
```
# Firewall Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/firewall
Google Cloud Firewall is a **network security feature** that lets you define rules to allow or block traffic to and from your VM instances.
This handler helps you **list firewall rules** in your GCP project using service account authentication.
It makes it easier to fetch firewall configurations for auditing, monitoring, or compliance checks.
## Example
To create the `GCPFirewallHandler` object with GCP credentials:
```python theme={null}
import os
from superagentx.handler.gcp.firewall import GCPFirewallHandler
# Initialize with a service account file
firewall_handler = GCPFirewallHandler(
creds="/path/to/service-account.json"
)
```
**List Firewall Rules:**
The get\_firewall\_details method retrieves all firewall rules in your GCP project.
It returns details like name, allowed protocols/ports, source/destination ranges, and priority.
```python theme={null}
firewalls = await firewall_handler.get_firewall_details()
for fw in firewalls:
print(fw.name, fw.direction, fw.priority)
```
# Gmail Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/gmail
The Gmail Handler enables seamless interaction with the Gmail API for sending, reading, and managing emails. It handles authentication using OAuth 2.0, and provides methods to fetch user profiles, read emails with attachments, send new emails, and create drafts. It is designed to integrate Gmail into automation workflows securely and efficiently.
## Example
To create the GmailHandler, initialize it with credentials JSON for OAuth:
```python theme={null}
from superagentx_handlers.gcp.gmail import GmailHandler
gmail_handler = GmailHandler(
credentials="credentials.json" # path to your OAuth 2.0 credentials file
)
```
**Get User Profile:**
Retrieve Gmail profile information for the authenticated user.
```python theme={null}
profile = await gmail_handler.get_user_profile()
print(profile)
```
**Read Emails:**
Fetches the latest emails, including sender, receiver, subject, body, and attachments.
```python theme={null}
emails = await gmail_handler.read_mail(max_results=10)
for mail in emails:
print(mail["subject"], mail["sender"])
```
**Send Email:**
Send an email via Gmail.
```python theme={null}
await gmail_handler.send_email(
from_address="you@example.com",
to="friend@example.com",
subject="Hello from GmailHandler",
content="This is a test email!"
)
```
**Create Draft Email:**
Create a draft email that can be sent later.
```python theme={null}
draft = await gmail_handler.create_draft_email(
from_address="you@example.com",
to="friend@example.com",
subject="Draft subject",
content="This is a saved draft message."
)
print(draft)
```
# IAM Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/iam
Google Cloud IAM (Identity and Access Management) allows you to securely manage access to your GCP resources.
It provides fine-grained controls for organizations, folders, and projects, ensuring that only authorized users and services can access resources.
With the **GCPIAMHandler**, you can fetch IAM policy details across organizations, folders, and projects, and also collect MFA-related evidence to strengthen compliance and security posture.
It helps with:
* Discovering IAM policies for GCP organizations, folders, and projects.
* Checking roles, bindings, and enforced conditions like MFA.
* Building a full compliance picture of your GCP environment.
## Example
### Initialization
To create a handler with your service account credentials:
```python theme={null}
from superagentx_handlers.gcp.iam import GCPIAMHandler
iam_handler = GCPIAMHandler(
creds="service_account.json", # path to service account json
scope=["https://www.googleapis.com/auth/cloud-platform"]
)
```
**Collect Organization IAM Evidence:**
Fetch IAM policy evidence for all accessible GCP organizations.
```python theme={null}
org_evidence = await iam_handler.collect_organization_iam_evidence()
print(org_evidence)
```
**Collect Folder IAM Evidence:**
Fetch IAM policy evidence for folders under a specific organization or folder.
```python theme={null}
folder_evidence = await iam_handler.collect_folder_iam_evidence(
parent_resource="organizations/123456789"
)
print(folder_evidence)
```
**Collect Project IAM Evidence:**
Fetch IAM policies for projects under an organization or folder.
```python theme={null}
project_evidence = await iam_handler.collect_project_iam_evidence(
parent_resource="folders/987654321"
)
print(project_evidence)
```
**Collect All IAM Evidence:**
Collects IAM evidence across organizations, folders, and projects in one run.
```python theme={null}
all_evidence = await iam_handler.collect_all_iam_evidence()
print(all_evidence)
```
**\_get\_resource\_iam\_policy(resource\_name: str, resource\_type: str):**
Internal method that retrieves IAM policy details for a specific resource (organization, folder, or project).
It returns information about roles, members, bindings, conditions, and MFA enforcement.
```python theme={null}
policy = await iam_handler._get_resource_iam_policy(
resource_name="projects/my-project-123",
resource_type="project"
)
print(policy)
```
# Load Balancer Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/loadbalancer
Google Cloud Load Balancer is a fully distributed, software-defined managed service for distributing traffic across multiple instances, services, and regions.
It supports HTTP(S), TCP/SSL, and UDP load balancing, integrates seamlessly with Google Cloud services, and provides **high availability, scalability, and global reach**.
The `GCPLoadBalancerHandler` provides programmatic access to **all load balancer components**, including URL maps, backend services, proxies, forwarding rules, SSL certificates, and health checks.
## Example
To create the `GCPLoadBalancerHandler` with Google Cloud credentials:
```python theme={null}
import os
from superagentx_handlers.gcp.load_balancer import GCPLoadBalancerHandler
# Initialize handler with a service account file
handler = GCPLoadBalancerHandler(
creds=os.getenv("GOOGLE_APPLICATION_CREDENTIALS")
)
```
**List All Load Balancer Components:**
Fetches all components (URL maps, backend services, proxies, SSL certs, health checks) concurrently.
```python theme={null}
components = await handler.list_all_load_balancer_components()
print(components.keys())
# Output: dict with url_maps, backend_services, proxies, ssl_certificates, health_checks
```
**Get URL Maps (Application Load Balancers):**
Retrieves URL maps including host rules, path matchers, and route actions.
```python theme={null}
url_maps = await handler.get_url_maps()
print(url_maps)
```
**Get Backend Services:**
Fetches backend services with protocols, ports, CDN settings, session affinity, and backend group details.
```python theme={null}
backends = await handler.get_backend_services()
print(backends)
```
**Get Target HTTP Proxies:**
Lists all HTTP proxies linked to URL maps.
```python theme={null}
http_proxies = await handler.get_target_http_proxies()
print(http_proxies)
```
**Get Target HTTPS Proxies:**
Lists all HTTPS proxies along with SSL certificates and QUIC override settings.
```python theme={null}
https_proxies = await handler.get_target_https_proxies()
print(https_proxies)
```
**Get Global Forwarding Rules:**
Fetches forwarding rules with IP addresses, protocols, port ranges, and targets.
```python theme={null}
rules = await handler.get_global_forwarding_rules()
print(rules)
```
**Get SSL Certificates:**
Retrieves SSL certificates (truncated for security) with SANs and expiration details.
```python theme={null}
certs = await handler.get_ssl_certificates()
print(certs)
```
**Get Health Checks:**
Lists health checks with thresholds, intervals, timeouts, and protocol-specific details (HTTP/HTTPS).
```python theme={null}
health_checks = await handler.get_health_checks()
print(health_checks)
```
# Security Role Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/security_role
The **GcpSecurityRoleHandler** is a specialized handler for Google Cloud Platform (GCP) that helps collect comprehensive **security and IAM-related information**.
It focuses on discovering and retrieving IAM policies, service accounts, and custom roles across **organizations, folders, and projects**.
This is particularly useful for compliance, governance, and security evidence collection across multiple levels of the GCP resource hierarchy.
## Example
To create the handler, pass in your GCP **service account credentials** (either as a JSON dict/string or use the environment variable `GOOGLE_APPLICATION_CREDENTIALS` pointing to your JSON key file).
```python theme={null}
import os
from superagentx_handlers.gcp.security_role import GcpSecurityRoleHandler
gcp_handler = GcpSecurityRoleHandler(
service_account_info=os.getenv("GCP_SERVICE_ACCOUNT_JSON")
)
```
**Collect Organization IAM Policies:**
Retrieves IAM policies for all accessible organizations.
```python theme={null}
org_policies = await gcp_handler.collect_organization_iam()
print(org_policies)
```
**Collect Folder IAM Policies:**
Fetches IAM policies for all folders under an organization or another folder.
```python theme={null}
folder_policies = await gcp_handler.collect_folder_iam(organization_id="1234567890")
print(folder_policies)
```
**Collect Project IAM Policies:**
Retrieves IAM policies for specific projects or all accessible projects.
```python theme={null}
project_policies = await gcp_handler.collect_project_iam(project_id="my-gcp-project")
print(project_policies)
```
**Collect Service Accounts:**
Lists service accounts in a given project.
```python theme={null}
service_accounts = await gcp_handler.collect_service_accounts(project_id="my-gcp-project")
print(service_accounts)
```
**Collect Custom Roles:**
Fetches custom IAM roles at the project or organization level.
```python theme={null}
custom_roles = await gcp_handler.collect_custom_roles(project_id="my-gcp-project")
print(custom_roles)
```
**Collect All Security Information:**
Performs a comprehensive security collection across IAM policies, service accounts, and custom roles for an organization or project.
```python theme={null}
security_info = await gcp_handler.collect_all_security_info(organization_id="1234567890")
print(security_info.keys())
```
# Storage Handler
Source: https://docs.superagentx.ai/examples/handlers/Gcp/storage
Google Cloud Storage (GCS) is a scalable, fully managed object storage service that lets you store and retrieve any amount of data at any time.
It supports storing unstructured data such as images, videos, backups, and datasets with **high durability, availability, and security**.
The **GCPStorageHandler** enables seamless interaction with Google Cloud Storage for tasks like managing buckets, listing files, retrieving metadata, and fetching IAM policies.
## Example
```python theme={null}
import os
from superagentx_handlers.gcp.storage import GCPStorageHandler
gcs_handler = GCPStorageHandler(
service_account_info=os.getenv("GCP_SERVICE_ACCOUNT_JSON")
)
```
**List Buckets:**
Retrieve all buckets in the Google Cloud Storage project.
```python theme={null}
buckets = await gcs_handler.list_buckets()
print(buckets)
```
**Get Bucket:**
Retrieve detailed information about a specific bucket.
```python theme={null}
bucket_info = await gcs_handler.get_bucket("my-bucket")
print(bucket_info)
```
**Get Bucket ACL:**
Retrieve the Access Control List (ACL) of a bucket.
```python theme={null}
acl = await gcs_handler.get_bucket_acl("my-bucket")
print(acl)
```
**Get Bucket CORS:**
Retrieve the CORS configuration for a bucket.
```python theme={null}
cors = await gcs_handler.get_bucket_cors("my-bucket")
print(cors)
```
**Get Bucket Encryption:**
Retrieve the default KMS encryption key of a bucket.
```python theme={null}
encryption = await gcs_handler.get_bucket_encryption("my-bucket")
print(encryption)
```
**Get Bucket Labels:**
Retrieve all labels assigned to a bucket.
```python theme={null}
labels = await gcs_handler.get_bucket_labels("my-bucket")
print(labels)
```
**Get Bucket Lifecycle Rules:**
Retrieve lifecycle rules applied to a bucket.
```python theme={null}
rules = await gcs_handler.get_bucket_lifecycle_rules("my-bucket")
print(rules)
```
**Get Bucket Location:**
Retrieve the geographic location of a bucket.
```python theme={null}
location = await gcs_handler.get_bucket_location("my-bucket")
print(location)
```
**Get Bucket Logging:**
Retrieve the logging configuration of a bucket.
```python theme={null}
logging_cfg = await gcs_handler.get_bucket_logging("my-bucket")
print(logging_cfg)
```
**Get Bucket Notifications:**
Retrieve the notification configurations of a bucket.
```python theme={null}
notifications = await gcs_handler.get_bucket_notification_config("my-bucket")
print(notifications)
```
**Get Bucket Owner:**
Retrieve the ownership information of a bucket.
```python theme={null}
owner = await gcs_handler.get_bucket_owner("my-bucket")
print(owner)
```
**Get Bucket IAM Policy:**
Retrieve the IAM policy attached to a bucket.
```python theme={null}
iam_policy = await gcs_handler.get_bucket_iam_policy("my-bucket")
print(iam_policy)
```
**Get All Buckets Info:**
Retrieve full details (ACL, encryption, IAM, logging, notifications) of all buckets.
```python theme={null}
all_info = await gcs_handler.get_all_buckets_info()
print(all_info)
```
**List Files:**
List all files (blobs) in a bucket with an optional prefix.
```python theme={null}
files = await gcs_handler.list_files("my-bucket", prefix="folder/")
print(files)
```
**Get File Info:**
Retrieve metadata and properties of a file (blob) in a bucket.
```python theme={null}
file_info = await gcs_handler.get_file_info(
bucket_name="my-bucket",
file_name="data/report.csv"
)
print(file_info)
```
# Calender Handler
Source: https://docs.superagentx.ai/examples/handlers/Google/calender
The Google Calendar Handler is a utility designed to simplify interactions with the Google Calendar API.
This handler enables users to authenticate, access, and manage their calendar events efficiently.
It provides methods for retrieving events for specific time frames, such as today, the upcoming week,
or the next month, as well as filtering events based on specific types.
## Example
A CalendarHandler for Google Calendar using provided credentials. This setup enables the application to manage calendar
events, schedule meetings, and interact with Google Calendar services programmatically.
To setup and get a credentials refer here.
After creating and enabling the Google Cloud API, you need to provide the path to the credentials in the *credentials* parameters.
```python calender_handler.py theme={null}
import asyncio
from superagentx_handlers.google.calender import CalenderHandler
async def get_today_events():
calender_handler = CalenderHandler(
credentials=""
)
return await calender_handler.get_today_events()
async def main():
res = await get_today_events()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```inline Calender Today Event theme={null}
{
"kind": "calendar#events",
"etag": "\"p32oftunmn22oe0o\"",
"summary": "sam1234@gmail.com",
"description": "",
"updated": "2024-07-01T08:41:06.398Z",
"timeZone": "America/Los_Angeles",
"accessRole": "owner",
"defaultReminders": [
{
"method": "popup",
"minutes": 10
}
],
"items": []
}
```
# Gmail Handler
Source: https://docs.superagentx.ai/examples/handlers/Google/gmail
The Gmail Handler is a specialized tool designed to interact with the Gmail API, enabling users to manage their
email accounts programmatically. This handler simplifies the process of accessing user profiles, sending emails,
and creating draft emails. It uses OAuth 2.0 for authentication and provides a user-friendly interface to perform
various email-related tasks asynchronously.
## Example
Set up a GmailHandler using the provided credentials, allowing the application to send, receive, and manage emails
through Gmail programmatically.
To setup and get a credentials refer here.
After creating and enabling the Google Cloud API, you need to provide the path to the credentials in the *credentials* parameters.
```python Gmail_handler.py theme={null}
import asyncio
from superagentx_handlers.google.gmail import GmailHandler
async def get_user_profile():
gmail_handler = GmailHandler(
credentials=""
)
return await gmail_handler.get_user_profile()
async def main():
res = await get_user_profile()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```python Gmail user profile theme={null}
{
'emailAddress': 'bala234@gmail.com',
'messagesTotal': 3678,
'threadsTotal': 4352,
'historyId': '985428'
}
```
# Asana Handler
Source: https://docs.superagentx.ai/examples/handlers/asana
Asana is a powerful project management platform designed to organize work,
track tasks, and streamline team collaboration. Through the Asana API, developers can programmatically
manage workspaces, users, projects, tasks, dependencies, and portfolios. The AsanaHandler provides a clean,
asynchronous interface for integrating Asana capabilities into agentic and automation-driven workflows.
## Example
To create the AsanaHandler object, initialize it with your Asana Personal Access Token (PAT).
The handler authenticates using the Asana API client and exposes core operations as tools.
```python theme={null}
import os
from superagentx_handlers.asana import AsanaHandler
asana_handler = AsanaHandler(
token=os.getenv("ASANA_ACCESS_TOKEN")
)
```
**Get Workspaces:**
Retrieves all Asana workspaces accessible to the authenticated user.
```python theme={null}
workspaces = await asana_handler.get_workspaces()
print(workspaces)
```
**Tasks:**
**Create Task:**
Creates a new task in a specified workspace with optional project, assignee, and due date.
```python theme={null}
task = await asana_handler.create_task(
name="Design API documentation",
workspace_id="123456789",
project_id="987654321",
assignee_id="1122334455",
due_on="2026-02-10"
)
print(task)
```
**Update Task:**
Updates task details such as name, completion status, assignee, or due date.
```python theme={null}
updated_task = await asana_handler.update_task(
task_id="123456789",
completed=True
)
print(updated_task)
```
**Delete Task:**
Deletes a task permanently from Asana.
```python theme={null}
response = await asana_handler.delete_task(
task_id="123456789"
)
print(response)
```
**Users:**
**Get Users:**
Retrieves all users across all accessible workspaces.
```python theme={null}
users = await asana_handler.get_users()
for user in users["data"]:
print(user["name"])
```
**Projects:**
**Get Projects:**
Fetches all projects across accessible workspaces.
```python theme={null}
projects = await asana_handler.get_projects()
print(projects)
```
**Task Dependencies:**
**Get Task Dependencies:**
Retrieves tasks that must be completed before a given task.
```python theme={null}
dependencies = await asana_handler.get_dependencies_for_task(
task_id="123456789"
)
print(dependencies)
```
**Get Task Dependents:**
Retrieves tasks that depend on the given task.
```python theme={null}
dependents = await asana_handler.get_dependents_for_task(
task_id="123456789"
)
print(dependents)
```
**Portfolios:**
**Get Portfolios:**
Retrieves all portfolios across accessible workspaces and users.
```python theme={null}
portfolios = await asana_handler.get_portfolios()
print(portfolios)
```
# Confluence Handler
Source: https://docs.superagentx.ai/examples/handlers/atlassian/confluence
A Confluence handler is a tool or API that allows developers to interact with Confluence,
a popular collaboration and documentation platform by Atlassian. This handler makes it easier
to manage and manipulate content within Confluence, such as creating, updating, and retrieving
pages, spaces, and user information.
## Example
This code connects to Confluence using stored credentials and retrieves all available spaces,
making it easy to access and manage content from the platform.
To set up Confluence with email, token, and organization access, you can follow these steps in the given link.
> Set up Email, Token and Organisation as an environmental variable and run the following code.
```python theme={null}
export ATLASSIAN_EMAIL = "**********"
export ATLASSIAN_TOKEN = "*************"
export ATLASSIAN_ORGANIZATION = "***************"
```
```python confluence_handler.py theme={null}
import os
import asyncio
from superagentx_handlers.atlassian.confluence import ConfluenceHandler
async def get_all_spaces():
confluence_handler = ConfluenceHandler(
email="",
token="",
organization=""
)
return await confluence_handler.get_all_spaces()
async def main():
res = await get_all_spaces()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
This JSON response provides detailed information about spaces in a Confluence instance,
including their IDs, names, types, status, and relevant links for accessing settings, content, and operations.
```python confluence theme={null}
{ '_links': { 'base': 'https://example.atlassian.net/wiki',
'context': '/wiki',
'self': 'https://example.atlassian.net/wiki/rest/api/space'},
'limit': 25,
'results': [ { '_expandable': { 'description': '',
'history': '',
'homepage': '/rest/api/content/74647464',
'icon': '',
'identifiers': '',
'lookAndFeel': '/rest/api/settings/lookandfeel?spaceKey=~74647464',
'metadata': '',
'operations': '',
'permissions': '',
'roles': '',
'settings': '/rest/api/space/~74647464/settings',
'theme': '/rest/api/space/~74647464/theme'},
'_links': { 'self': 'https://example.atlassian.net/wiki/rest/api/space/~74647464',
'webui': '/spaces/~74647464'},
'id': 98765432,
'key': '~2345678',
'name': 'Joe',
'status': 'current',
'type': 'personal'},
{ '_expandable': { 'description': '',
'history': '',
'homepage': '/rest/api/content/765432',
'icon': '',
'identifiers': '',
'lookAndFeel': '/rest/api/settings/lookandfeel?spaceKey=~657482927',
'metadata': '',
'operations': '',
'permissions': '',
'roles': '',
'settings': '/rest/api/space/~657482927/settings',
'theme': '/rest/api/space/~657482927/theme'},
'_links': { 'self': 'https://example.atlassian.net/wiki/rest/api/space/~657482927',
'webui': '/spaces/~657482927'},
'id': 171507714,
'key': '~74647464',
'name': 'joe',
'status': 'current',
'type': 'personal'},
{ '_expandable': { 'description': '',
'history': '',
'homepage': '/rest/api/content/74647464',
'icon': '',
'identifiers': '',
'lookAndFeel': '/rest/api/settings/lookandfeel?spaceKey=~74647464',
'metadata': '',
'operations': '',
'permissions': '',
'roles': '',
'settings': '/rest/api/space/~74647464/settings',
'theme': '/rest/api/space/~74647464/theme'},
'_links': { 'self': 'https://example.atlassian.net/wiki/rest/api/space/DPS',
'webui': '/spaces/dgf'},
'id': 965443593,
'key': 'dgf',
'name': 'Example Product Sprints',
'status': 'current',
'type': 'global'}],
'size': 24,
'start': 0}
```
# Jira Handler
Source: https://docs.superagentx.ai/examples/handlers/atlassian/jira
The JiraHandler class is a tool that helps developers easily connect to Jira and perform common
operations like creating, updating, and managing issues or projects. If youโve ever used Jira
for tracking bugs, managing tasks, or following project progress, this class makes it simpler
to automate those activities using Python.
## Jira Handler Code
The JiraHandler connects to Jira using an email, API token, and organization details. This setup allows the program to
interact with Jira, making it easy to manage tasks, projects, and issues programmatically.
To set up Confluence with email, token, and organization access, you can follow these steps in the given link.
> Set up Email, Token and Organisation as an environmental variable and run the following code.
```python theme={null}
export ATLASSIAN_EMAIL = "**********"
export ATLASSIAN_TOKEN = "*************"
export ATLASSIAN_ORGANIZATION = "***************"
```
```python jira_handler.py theme={null}
import os
import asyncio
from superagentx_handlers.atlassian.jira import JiraHandler
async def get_active_sprint():
jira_handler = JiraHandler(
email="",
token="",
organization=""
)
return await jira_handler.get_active_sprint(
board_id=1,
start=0,
end=50
)
async def main():
res = await get_active_sprint()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```python jira theme={null}
[]
```
# API Gateway Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/api_gateway
Amazon API Gateway is a fully managed service that makes it easy for developers to create, publish, maintain,
monitor, and secure APIs at any scale.
The `AWSAPIGatewayHandler` class provides async methods to manage **REST APIs, resources, methods, stages, deployments, usage plans, and API keys**.
## **Initialization**
```python apigateway_handler_test.py theme={null}
import os
from superagentx_handlers.aws.api_gateway import AWSAPIGatewayHandler
api_handler = AWSAPIGatewayHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="eu-central-1"
)
```
**Get All API Gateways Info:**
Fetches a summary of all API Gateways (REST, HTTP, and WebSocket APIs) with their resources, stages, and deployments.
```python theme={null}
apis = await api_handler.get_all_api_gateways_info()
print(apis)
```
**Get REST APIs (API Gateway v1):**
Retrieves a list of all REST APIs in your account.
```python theme={null}
rest_apis = await api_handler.get_rest_apis()
print(rest_apis)
```
**Get Resources for a REST API:**
Lists all resources (paths/endpoints) for a given REST API.
```python theme={null}
resources = await api_handler.get_rest_api_resources(api_id="your_rest_api_id")
print(resources)
```
**Get Stages for a REST API:**
Returns all stages (e.g., dev, prod) configured for a REST API.
```python theme={null}
stages = await api_handler.get_rest_api_stages(api_id="your_rest_api_id")
print(stages)
```
**Get Deployments for a REST API:**
Lists deployments of a given REST API.
```python theme={null}
deployments = await api_handler.get_rest_api_deployments(api_id="your_rest_api_id")
print(deployments)
```
**Get HTTP APIs (API Gateway v2):**
Retrieves all HTTP APIs in your account.
```python theme={null}
http_apis = await api_handler.get_http_apis()
print(http_apis)
```
**Get Routes for an HTTP API:**
Lists all routes (endpoints) for an HTTP API.
```python theme={null}
routes = await api_handler.get_http_api_routes(api_id="your_http_api_id")
print(routes)
```
**Get Stages for an HTTP API:**
Returns all stages (e.g., beta, prod) for an HTTP API.
```python theme={null}
stages = await api_handler.get_http_api_stages(api_id="your_http_api_id")
print(stages)
```
**Get Integrations for an HTTP API:**
Lists integrations (like Lambda, VPC Link, etc.) for an HTTP API.
```python theme={null}
integrations = await api_handler.get_http_api_integrations(api_id="your_http_api_id")
print(integrations)
```
**Get WebSocket APIs (API Gateway v2):**
Retrieves all WebSocket APIs in your account.
```python theme={null}
ws_apis = await api_handler.get_websocket_apis()
print(ws_apis)
```
**Get Routes for a WebSocket API:**
Lists all routes for a WebSocket API.
```python theme={null}
routes = await api_handler.get_websocket_api_routes(api_id="your_ws_api_id")
print(routes)
```
**Get Stages for a WebSocket API:**
Fetches all stages for a WebSocket API.
```python theme={null}
stages = await api_handler.get_websocket_api_stages(api_id="your_ws_api_id")
print(stages)
```
**Get Integrations for a WebSocket API:**
Lists integrations for a WebSocket API.
```python theme={null}
integrations = await api_handler.get_websocket_api_integrations(api_id="your_ws_api_id")
print(integrations)
```
**Get VPC Links (REST APIs v1):**
Fetches all VPC Links configured for REST APIs.
```python theme={null}
vpc_links = await api_handler.get_vpc_links_v1()
print(vpc_links)
```
**Get VPC Links (HTTP/WebSocket APIs v2):**
Fetches all VPC Links configured for HTTP and WebSocket APIs.
```python theme={null}
vpc_links = await api_handler.get_vpc_links_v2()
print(vpc_links)
```
# CloudFront Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/cloudfront
Amazon CloudFront is a fast Content Delivery Network (CDN) service that securely delivers data, videos, applications,
and APIs to customers globally with low latency and high transfer speeds. CloudFront works seamlessly with services
like S3, API Gateway, and ACM for SSL/TLS certificates.
## Example
To create the AWSCloudFrontHandler object with AWS credentials.
The AWSCloudFrontHandler enables you to interact with CloudFront for tasks such as
listing distributions, cache behaviors, access control settings, and certificates.
```python cloudfront_handler_test.py theme={null}
import os
from superagentx_handlers.aws.cloudfront import AWSCloudFrontHandler
cloudfront_handler = AWSCloudFrontHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**List Distributions:**
The list\_distributions method retrieves all CloudFront distributions.
```python theme={null}
distributions = await cloudfront_handler.list_distributions()
print(distributions)
```
**List Certificates:**
The list\_certificates method retrieves all ACM-managed certificates in us-east-1,
typically used with CloudFront.
```python theme={null}
certificates = await cloudfront_handler.list_certificates()
print(certificates)
```
**List Cache Behaviors:**
The list\_cache\_behaviors method shows default and ordered cache behaviors for all distributions.
```python theme={null}
cache_behaviors = await cloudfront_handler.list_cache_behaviors()
print(cache_behaviors)
```
**List Access Control Configurations:**
The list\_access\_control\_configs method retrieves access control settings like geo-restrictions and WAF associations.
```python theme={null}
access_controls = await cloudfront_handler.list_access_control_configs()
print(access_controls)
```
# CloudWatch Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/cloudwatch
Amazon CloudWatch is a monitoring and observability service that provides data and actionable insights for AWS, hybrid,
and on-premises applications and infrastructure resources. It allows you to collect metrics, logs, and events, helping
you monitor system health, detect anomalies, and keep applications running smoothly.
## Example
To create the AWSCloudWatchHandler object with AWS credentials.
The AWSCloudWatchHandler connects to Amazon CloudWatch using access credentials
(access key, secret key) and specifies a region. It enables seamless interaction
with CloudWatch for tasks like fetching metrics and monitoring resources.
```python cloudwatch_handler_test.py theme={null}
import os
from superagentx_handlers.aws.cloudwatch import AWSCloudWatchHandler
cloudwatch_handler = AWSCloudWatchHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**Get Metric Data:**
The get\_metric\_data method retrieves metrics for a specific namespace, metric, and dimensions within a given time range.
from datetime import datetime, timedelta
```python theme={null}
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=1)
response = await cloudwatch_handler.get_metric_data(
namespace="AWS/EC2",
metric_name="CPUUtilization",
start_time=start_time,
end_time=end_time,
period=300, # granularity in seconds
statistics=["Average"], # can also use Sum, Minimum, Maximum, etc.
dimensions=[{"Name": "InstanceId", "Value": "i-1234567890abcdef0"}]
)
print(response)
```
# AWS Cognito Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/cognito
AWS Cognito is a fully managed identity service that provides user authentication, authorization, and user
management for modern applications. It supports user pools, identity providers (IdPs), MFA, OAuth flows,
and fine-grained access control.
The AWSCognitoHandler provides an asynchronous, agent-friendly interface to the AWS Cognito Identity Provider (IDP)
API. It is designed for automation, governance, and security workflows, enabling programmatic control over users,
groups, authentication flows, MFA, and user pool configuration.
The handler supports secure credential management using AWS STS role assumption and exposes Cognito operations
as LLM-compatible tools.
## Example
To create the AWSCognitoHandler, initialize it with AWS credentials or an IAM role.
The handler automatically assumes the role (if provided) and initializes a Cognito client.
```python theme={null}
import os
from superagentx_handlers.aws.cognito import AWSCognitoHandler
cognito_handler = AWSCognitoHandler(
region_name=os.getenv("AWS_REGION"),
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
aws_role_arn=os.getenv("AWS_ROLE_ARN"),
role_session_name="superagentx-session"
)
```
## User Pool Management:
**List User Pools:**
Retrieves all Cognito user pools in the AWS account.
```python theme={null}
pools = await cognito_handler.list_user_pools()
print(pools)
```
**Create User Pool:**
Creates a new user pool with optional policies and schema.
```python theme={null}
pool = await cognito_handler.create_user_pool(
pool_name="my-user-pool"
)
print(pool)
```
**Describe User Pool:**
Fetches metadata and configuration of a specific user pool.
```python theme={null}
details = await cognito_handler.describe_user_pool(
user_pool_id="us-east-1_XXXXXXX"
)
print(details)
```
## User Pool Management:
**Create User (Admin):**
Creates a new user with a temporary password.
```python theme={null}
user = await cognito_handler.admin_create_user(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe",
temporary_password="Temp@123",
user_attributes=[
{"Name": "email", "Value": "john@example.com"}
]
)
print(user)
```
**Get User Details:**
Retrieves user attributes and status.
```python theme={null}
user = await cognito_handler.admin_get_user(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe"
)
print(user)
```
**Enable / Disable User:**
Controls user access within the pool.
```python theme={null}
await cognito_handler.admin_disable_user(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe"
)
await cognito_handler.admin_enable_user(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe"
)
```
**Delete User:**
Removes a user permanently.
```python theme={null}
await cognito_handler.admin_delete_user(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe"
)
```
## Group Management:
**Create Group:**
Creates a new group in a user pool.
```python theme={null}
group = await cognito_handler.create_group(
user_pool_id="us-east-1_XXXXXXX",
group_name="admins",
description="Administrator group"
)
print(group)
```
**Add User to Group:**
Assigns a user to a group.
```python theme={null}
await cognito_handler.admin_add_user_to_group(
user_pool_id="us-east-1_XXXXXXX",
username="john.doe",
group_name="admins"
)
```
**List Groups:**
Lists all groups in a user pool.
```python theme={null}
groups = await cognito_handler.list_groups(
user_pool_id="us-east-1_XXXXXXX"
)
print(groups)
```
## Authentication & Authorization:
**Initiate Authentication:**
Starts an authentication flow (USER\_PASSWORD\_AUTH, REFRESH\_TOKEN\_AUTH, etc.).
```python theme={null}
auth = await cognito_handler.initiate_auth(
client_id="CLIENT_ID",
auth_flow="USER_PASSWORD_AUTH",
auth_parameters={
"USERNAME": "john.doe",
"PASSWORD": "Password@123"
}
)
print(auth)
```
**Respond to Auth Challenge:**
Handles MFA or password challenges.
```python theme={null}
challenge = await cognito_handler.respond_to_auth_challenge(
client_id="CLIENT_ID",
challenge_name="SOFTWARE_TOKEN_MFA",
challenge_responses={
"USERNAME": "john.doe",
"SOFTWARE_TOKEN_MFA_CODE": "123456"
},
session=auth.get("Session")
)
print(challenge)
```
**Global Sign-Out:**
Signs the user out from all active sessions.
```python theme={null}
await cognito_handler.global_sign_out(
access_token="ACCESS_TOKEN"
)
```
## MFA & Security:
**Set User MFA Preference:**
Configures MFA for a user.
```python theme={null}
await cognito_handler.set_user_mfa_preference(
access_token="ACCESS_TOKEN",
software_mfa={"Enabled": True, "PreferredMfa": True}
)
```
**Verify Software Token:**
Verifies an MFA token.
```python theme={null}
await cognito_handler.verify_software_token(
access_token="ACCESS_TOKEN",
user_code="123456"
)
```
## Identity Providers (IdP)
**Create Identity Provider:**
Adds an external IdP (Google, SAML, OIDC).
```python theme={null}
idp = await cognito_handler.create_identity_provider(
user_pool_id="us-east-1_XXXXXXX",
provider_name="Google",
provider_type="Google",
provider_details={
"client_id": "GOOGLE_CLIENT_ID",
"client_secret": "GOOGLE_SECRET"
}
)
print(idp)
```
**List Identity Providers:**
Lists all configured IdPs.
```python theme={null}
providers = await cognito_handler.list_identity_providers(
user_pool_id="us-east-1_XXXXXXX"
)
print(providers)
```
## Session & Resource Management:
**Tag Resources:**
Adds tags to a Cognito resource.
```python theme={null}
await cognito_handler.tag_resource(
resource_arn="RESOURCE_ARN",
tags={"env": "prod", "team": "security"}
)
```
**Close Handler:**
Gracefully releases underlying AWS resources.
```python theme={null}
await cognito_handler.close()
```
# EC2 Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/ec2
Amazon EC2 (Elastic Compute Cloud) is a service that provides secure, resizable compute capacity in the cloud.
It allows you to launch virtual servers (instances), manage storage volumes, security groups, and networking configurations.
With the EC2 handler, you can programmatically retrieve and manage EC2 resources such as instances, volumes, AMIs, snapshots, and more.
## Example
To create the AWSEC2Handler object with AWS credentials.
The AWSEC2Handler connects to AWS EC2 using access credentials (access key, secret key) and a region name.
It enables seamless interaction with EC2 for retrieving and managing compute resources.
```python ec2_handler_test.py theme={null}
import os
from superagentx_handlers.aws.ec2 import AWSEC2Handler
ec2_handler = AWSEC2Handler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="eu-central-1"
)
```
**Get Instances:**
The get\_instances method fetches all EC2 instances.
```python theme={null}
await ec2_handler.get_instances()
print(instances)
```
**Get Security Groups:**
The get\_security\_groups method fetches all security groups.
```python theme={null}
await ec2_handler.get_security_groups()
print(security_groups)
```
**Get Volumes:**
The get\_volumes method fetches all EBS volumes.
```python theme={null}
await ec2_handler.get_volumes()
print(volumes)
```
**Get AMIs:**
The get\_amis method fetches all AMIs owned by the account.
```python theme={null}
await ec2_handler.get_amis()
print(amis)
```
**Get Snapshots:**
The get\_snapshots method fetches all snapshots owned by the account.
```python theme={null}
await ec2_handler.get_snapshots()
print(snapshots)
```
**Get Key Pairs:**
The get\_key\_pairs method fetches all key pairs.
```python theme={null}
key_pairs = await ec2_handler.get_key_pairs()
print(key_pairs)
```
**Get Network Interfaces:**
The get\_network\_interfaces method fetches all network interfaces.
```python theme={null}
network_interfaces = await ec2_handler.get_network_interfaces()
print(network_interfaces)
```
**Collect All EC2 Resources:**
The collect\_all\_ec2 method fetches all EC2 resources in parallel.
```python theme={null}
await ec2_handler.collect_all_ec2()
print(all_ec2_data)
```
# ECS Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/ecs
Amazon ECS (Elastic Container Service) is a fully managed container orchestration service.
It allows you to run Docker containers at scale, with integration to EC2 and Fargate.
The ECS Handler helps fetch **clusters, services, tasks, and container instances**.
## Example
To create the **ECS Handler**, pass AWS credentials (`access_key`, `secret_key`, and `region`).
```python theme={null}
import os
from superagentx_handlers.aws.ecs import AWSECSHandler
ecs_handler = AWSECSHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="eu-central-1"
)
```
**List Clusters:**
Fetches all ECS clusters.
```python theme={null}
clusters = await ecs_handler.list_clusters()
print(clusters)
```
**List Services:**
Lists ECS services running inside a cluster.
```python theme={null}
services = await ecs_handler.list_services(cluster="my-cluster")
print(services)
```
**List Tasks:**
Lists running tasks inside a service.
```python theme={null}
tasks = await ecs_handler.list_tasks(cluster="my-cluster")
print(tasks)
```
**Describe Cluster:**
Fetches detailed information about a cluster.
```python theme={null}
cluster_info = await ecs_handler.describe_cluster(cluster="my-cluster")
print(cluster_info)
```
**Describe Service:**
Fetches detailed info of a service inside a cluster.
```python theme={null}
service_info = await ecs_handler.describe_service(
cluster="my-cluster",
service="my-service"
)
print(service_info)
```
**Describe Task:**
Fetches details about a specific task.
```python theme={null}
task_info = await ecs_handler.describe_task(
cluster="my-cluster",
task="0123456789abcdef0"
)
print(task_info)
```
# elasticache handler
Source: https://docs.superagentx.ai/examples/handlers/aws/elasticache
Amazon ElastiCache is a fully managed, in-memory caching service by AWS that supports Redis and Memcached engines.
It helps improve application performance by reducing database load and delivering data with microsecond latency.
With high availability, clustering, and automatic failover, ElastiCache integrates seamlessly with AWS applications.
## Example
To create the **AWSElasticCacheHandler**, provide your AWS credentials (access key, secret key) and region.
The handler allows you to retrieve details of ElastiCache clusters, replication groups, subnet groups, security groups,
VPCs, and serverless cache configurations.
```python theme={null}
# elasticache_handler_test.py
import os
from superagentx_handlers.aws.elasticache import AWSElasticCacheHandler
elasticache_handler = AWSElasticCacheHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**Get ElastiCache Details:**
Aggregates clusters, replication groups, subnet groups, security groups, and VPCs into one JSON summary.
```python theme={null}
details = await elasticache_handler.get_elastic_cache_details()
print(details)
```
**Get All ElastiCache Clusters:**
Fetches metadata for all clusters in the account/region.
```python theme={null}
clusters = await elasticache_handler.get_clusters()
print(clusters)
```
**Get All ElastiCache Replication Groups:**
Retrieves replication group information including member clusters and status.
```python theme={null}
replication_groups = await elasticache_handler.get_replication_groups()
print(replication_groups)
```
**Get All Cache Subnet Groups:**
Retrieves subnet groups and maps them to VPCs.
```python theme={null}
subnet_groups = await elasticache_handler.get_cache_subnet_groups()
print(subnet_groups)
```
**Update Clusters & Replication Groups with VPC Info:**
Enhances cluster and replication group objects with VPC IDs from subnet groups.
```python theme={null}
await elasticache_handler.update_vpc_info(clusters, replication_groups, subnet_groups)
```
**Get Unique Security Group IDs:**
Extracts a unique list of security group IDs from clusters and replication groups.
```python theme={null}
sg_ids = await elasticache_handler.get_unique_security_group_ids(clusters, replication_groups)
print(sg_ids)
```
**Get Security Group Details:**
Retrieves details for each security group (name, rules, VPC association).
```python theme={null}
sg_details = await elasticache_handler.get_security_groups(security_group_ids=["sg-0123456789abcdef"])
print(sg_details)
```
**Process Inbound Rules:**
Parses inbound (ingress) security group rules into a normalized JSON format.
```python theme={null}
inbound_rules = elasticache_handler.process_inbound_rules(sg["IpPermissions"])
print(inbound_rules)
```
**Process Outbound Rules:**
Parses outbound (egress) security group rules into a normalized JSON format.
```python theme={null}
outbound_rules = elasticache_handler.process_outbound_rules(sg["IpPermissionsEgress"])
print(outbound_rules)
```
**Get Unique VPC IDs:**
Extracts unique VPC IDs from clusters, replication groups, and subnet groups.
```python theme={null}
vpc_ids = await elasticache_handler.get_unique_vpc_ids(clusters, replication_groups, subnet_groups)
print(vpc_ids)
```
**Get VPC Details:**
Retrieves details for each VPC (CIDR, state, default flag).
```python theme={null}
vpc_details = await elasticache_handler.get_vpcs(vpc_ids=["vpc-1234567890abcdef"])
print(vpc_details)
```
# ELB Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/elb
Amazon Elastic Load Balancing (ELBv2) automatically distributes incoming application traffic across multiple targets, such as Amazon EC2 instances, containers, and IP addresses.
It provides high availability, automatic scaling, and robust security features, and supports multiple load balancer types (Application, Network, and Gateway).
This handler focuses on **Application Load Balancers (ALBs)** and retrieves details about load balancers, listeners, listener rules, target groups, and target health.
## Example
To create the 'AWSElasticLoadBalancerHandler' object, you must provide AWS credentials:
```python elb_handler_test.py theme={null}
import os
from superagentx_handlers.aws.elb import AWSElasticLoadBalancerHandler
elb_handler = AWSElasticLoadBalancerHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**Get ELB Details:**
Fetches all Application Load Balancer (ALB) details, including listeners, rules, target groups, and target health.
```python theme={null}
elb_details = await elb_handler.get_elb_details()
print(elb_details)
```
**Collect Target Groups from Default Actions and Rules:**
Collects all target groups referenced in listenersโ default actions and custom rules.
```python theme={null}
target_groups = await elb_handler.collect_target_groups_from_listeners("arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/app/my-alb/abc123")
print(target_groups)
```
**Fetch Detailed Target Group Info:**
Fetches complete target group configuration and health status, including protocol, port, health check config, and registered targets.
```python theme={null}
tg_info = await elb_handler.get_target_group_info("arn:aws:elasticloadbalancing:us-east-1:123456789012:targetgroup/my-tg/ghi789")
print(tg_info)
```
# AWS Helper
Source: https://docs.superagentx.ai/examples/handlers/aws/helper
The **AWS Helper** provides utility functions to simplify authentication and temporary credential generation.
One key feature is the ability to generate **AWS STS temporary credentials**, which are short-lived credentials used for secure authentication in AWS services without exposing long-term access keys.
STS (Security Token Service) is particularly useful when granting temporary access to applications, scripts, or external systems.
## Example
To use the AWS Helper, call the `generate_aws_sts_token` method with your AWS credentials and region details.
```python theme={null}
import os
from helpers.aws_helper import generate_aws_sts_token
# Generate temporary STS token
sts_credentials = generate_aws_sts_token(
region_name="us-east-1",
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
duration_seconds=3600 # 1 hour
)
print(sts_credentials)
```
# IAM Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/iam
Amazon **IAM (Identity and Access Management)** is a secure AWS service that enables you to manage access to AWS resources.
It allows you to create users, groups, roles, and policies to define permissions. IAM helps enforce the **principle of least privilege** by giving the right access to the right people or systems.
The **AWSIAMHandler** provides asynchronous helper methods to list and audit IAM entities, including users, groups, roles, managed policies, MFA devices, and account-level security configurations.
## Example
To create the **AWSIAMHandler** object, initialize it with AWS credentials and a region:
```python theme={null}
import os
from superagentx_handlers.aws.iam import AWSIAMHandler
iam_handler = AWSIAMHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**List IAM Users with Details:**
Fetches all IAM users with attached/inline policies, MFA, login profiles, access keys, and groups.
```python theme={null}
users = await iam_handler.list_iam_users_with_details()
print(users)
```
**List IAM Roles with Details:**
Fetches IAM roles with trust policies, attached policies, inline policies, and instance profiles.
```python theme={null}
roles = await iam_handler.list_iam_roles_with_details()
print(roles)
```
**List IAM Groups with Details:**
Lists IAM groups with members, attached managed policies, and inline policies.
```python theme={null}
groups = await iam_handler.list_iam_groups_with_details()
print(groups)
```
**List IAM Managed Policies:**
Retrieves all IAM managed policies (AWS & customer-managed) along with their documents.
```python theme={null}
policies = await iam_handler.list_iam_managed_policies_with_documents(scope="All")
print(policies)
```
**List MFA Enabled Users:**
Returns IAM users with MFA devices enabled.
```python theme={null}
mfa_users = await iam_handler.list_mfa_enabled_users()
print(mfa_users)
```
**Get Account Summary:**
Fetches IAM resource usage and quota (roles, groups, users, etc.).
```python theme={null}
summary = await iam_handler.get_account_summary()
print(summary)
```
**Get Credential Report:**
Retrieves IAM account credential report (CSV format) for auditing password age, MFA, and access key rotation.
```python theme={null}
report = await iam_handler.get_credential_report()
print(report)
```
**List Virtual MFA Devices:**
Lists assigned and unassigned virtual MFA devices.
```python theme={null}
mfa_devices = await iam_handler.list_virtual_mfa_devices(assignment_status="Any")
print(mfa_devices)
```
**Get Account Password Policy:**
Retrieves password policy (complexity, expiry, rotation, etc.).
```python theme={null}
policy = await iam_handler.get_account_password_policy()
print(policy)
```
**List Account Aliases:**
Lists aliases for the AWS account.
```python theme={null}
aliases = await iam_handler.list_account_aliases()
print(aliases)
```
**List Organization Accounts:**
Lists all accounts under AWS Organizations (if enabled).
```python theme={null}
accounts = await iam_handler.list_organization_accounts()
print(accounts)
```
**Collect All IAM Data:**
Fetches everything (users, groups, roles, policies, MFA, org accounts, etc.) in a single structured call.
```python theme={null}
iam_data = await iam_handler.collect_all_iam()
print(iam_data.keys())
```
# RDS Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/rds
Amazon RDS (Relational Database Service) is a managed service that makes it easy to set up, operate, and scale relational databases in the cloud.
It supports multiple engines like MySQL, PostgreSQL, MariaDB, Oracle, and SQL Server, and provides features such as automated backups, scaling, monitoring, and high availability.
## Example
To create the **RDS Handler**, you must pass AWS credentials (`access_key`, `secret_key`, and `region`).
This enables you to interact with RDS resources, clusters, proxies, and their associations with EC2.
```python theme={null}
import os
from superagentx_handlers.aws.rds import AWSRDSHandler
rds_handler = AWSRDSHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="eu-central-1"
)
```
**Get RDS Association Details:**
Provides a comprehensive inventory of RDS resources and their associations.
```python theme={null}
details = await rds_handler.get_rds_association_details()
print(details)
```
**List RDS Instances:**
Fetches all RDS database instances in the account.
```python theme={null}
instances = await rds_handler.get_rds_instances()
print(instances)
```
**List RDS Clusters:**
Fetches all Aurora RDS clusters.
```python theme={null}
clusters = await rds_handler.get_rds_clusters()
print(clusters)
```
**List RDS Proxies:**
Fetches all RDS database proxies.
```python theme={null}
proxies = await rds_handler.get_rds_proxies()
print(proxies)
```
**Get Proxy Targets:**
Retrieves targets associated with a specific RDS proxy.
```python theme={null}
targets = await rds_handler.get_proxy_targets(
proxy_name="my-rds-proxy"
)
print(targets)
```
**Get EC2 Associations:**
Finds EC2 instances that are associated with RDS security groups.
```python theme={null}
associations = await rds_handler.get_ec2_associations(
rds_instances=instances
)
print(associations)
```
**Get Backup Configuration:**
Fetches backup details such as retention period, backup window, and maintenance window for an RDS instance or cluster.
```python theme={null}
backup = await rds_handler.get_backup_config(
db_identifier="my-db-instance",
is_cluster=False # set True for clusters
)
print(backup)
```
# S3 Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/s3
Amazon S3 is a cloud storage service on AWS that lets you store different types of files, like photos, audio, and
videos, as objects. It offers high scalability and security, making it easy to store and access any amount of data
from anywhere, anytime. S3 also provides features like high availability, strong security, and smooth integration with
other AWS services.
## Example
To create the AWSHandler object creation with the aws credentials.
The AWSS3Handler connects to an Amazon S3 bucket using access credentials
(access key, secret key) and specifies a region and bucket name. It enables
seamless interaction with S3 for tasks like uploading, downloading, and
managing files.
```python s3_handler_test.py theme={null}
from superagentx_handlers.aws.s3 import AWSS3Handler
s3_handler = AWSS3Handler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
bucket_name="test",
region_name="eu-central-1"
)
```
**File Upload:**
The upload\_file method accepts a file name and an object name.
```python theme={null}
await s3_handler.upload_file(
file_name="test.txt",
object_name="test.txt"
)
```
**List Buckets:**
Returns a list of all buckets owned by the authenticated sender of the request.
```python theme={null}
buckets = await s3_handler.list_bucket()
print(buckets)
```
**Download File:**
The Download\_file method accepts a file download in the buckets
```python theme={null}
await s3_handler.download_file(
file_name="test.txt",
object_name="test.txt"
)
```
**List Objects in a Bucket:**
The List object method shows list all object in the bucket
```python theme={null}
await s3_handler.list_objects(prefix="folder/")
print(objects)
```
**Delete an Object:**
The Delete object can delete the object in the bucket
```python theme={null}
await s3_handler.delete_object(
object_name="test.txt"
)
```
**Generate a Pre-Signed URL:**
The method will generate\_presigned\_url for the account
```python theme={null}
await s3_handler.generate_presigned_url(
object_name="test.txt",
expiration=3600 # URL valid for 1 hour
)
print("Presigned URL:", url)
```
# Security Groups Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/security_groups
Amazon EC2 **Security Groups** act as virtual firewalls to control inbound and outbound traffic for your AWS resources, such as EC2 instances and load balancers. They help secure your applications by defining rules that allow or deny specific traffic. Security Groups operate at the instance level and are an essential part of AWS networking and security best practices.
This handler enables you to interact with **AWS Security Groups** using boto3, making it easier to retrieve details such as rules, attached instances, and associated VPCs. It provides an abstraction for developers to collect security group data seamlessly.
## Example
To create the AWS Security Groups Handler object with AWS credentials:
```python theme={null}
import os
from superagentx_handlers.aws.security_groups import AWSSecurityGroupsHandler
sg_handler = AWSSecurityGroupsHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="us-east-1"
)
```
**List Security Groups:**
Retrieves details of all security groups in your AWS account for the given region, including rules, VPCs, and attached resources.
```python theme={null}
security_groups = await sg_handler.get_ec2_security_groups()
print(security_groups)
```
# Serverless Handler
Source: https://docs.superagentx.ai/examples/handlers/aws/serverless
AWS Lambda is a **serverless compute service** that lets you run code without provisioning or managing servers. You just upload your code (as a ZIP package or container image), and Lambda automatically scales and runs it in response to events such as API Gateway requests, S3 object uploads, DynamoDB streams, or scheduled CloudWatch events.
Lambda integrates seamlessly with **VPC, IAM, Security Groups, and other AWS services**, making it a core component of modern serverless applications.
## Example
To create the AWSLambdaHandler with AWS credentials:
```python theme={null}
import os
from superagentx_handlers.aws.serverless import AWSLambdaHandler
lambda_handler = AWSLambdaHandler(
aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
region_name="eu-central-1"
)
```
**List Lambda Functions:**
Retrieves all Lambda functions in the account, including runtime, handler, role, security groups, IAM policies, and environment configurations.
```python theme={null}
functions = await lambda_handler.get_lambda_functions()
print(functions)
```
**Get IAM Role Policies for a Lambda Role:**
Fetches all attached and inline IAM policies for the IAM role associated with a Lambda.
```python theme={null}
policies = await lambda_handler.get_role_policies("MyLambdaRole")
print(policies)
```
**Get Security Group Details:**
Fetches inbound/outbound rules for security groups attached to a Lambda functionโs VPC.
```python theme={null}
sg_details = await lambda_handler.get_security_group_details(["sg-12345678"])
print(sg_details)
```
# Application Gateway
Source: https://docs.superagentx.ai/examples/handlers/azure/application_gateway
Azure Application Gateway is a web traffic load balancer that enables you to manage traffic to your web applications. It provides features like SSL termination, cookie-based session affinity, URL-based routing, and Web Application Firewall (WAF) integration to protect against common exploits and vulnerabilities. With Application Gateway, you can secure, scale, and monitor your application traffic efficiently in Azure.
## Example
To create the AzureApplicationGatewayHandler object with Azure credentials. The handler connects to Azure Application Gateway resources using client credentials or managed identity, and provides methods to fetch gateways, backend health, WAF policies, and networking components.
```python theme={null}
import os
from superagentx_handlers.azure.application_gateway import AzureApplicationGatewayHandler
gateway_handler = AzureApplicationGatewayHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
)
```
**Get Resource Groups:**
Retrieve all resource groups in the subscription.
```python theme={null}
groups = await gateway_handler.get_resource_groups()
print(groups)
```
**List Application Gateways:**
Fetch all Application Gateways in the subscription or within a resource group.
```python theme={null}
app_gateways = await gateway_handler.get_application_gateways(resource_group_name="my-rg")
print(app_gateways)
```
**Get Application Gateway Details:**
Fetch detailed info for a specific Application Gateway.
```python theme={null}
gateway = await gateway_handler.get_application_gateway(resource_group_name="my-rg", gateway_name="my-gateway")
print(gateway)
```
**Check Backend Health:**
Retrieve backend pool health information for a gateway.
```python theme={null}
health = await gateway_handler.get_application_gateway_backend_health(resource_group_name="my-rg", gateway_name="my-gateway")
print(health)
```
**Get WAF Policies:**
List all WAF policies associated with Application Gateways.
```python theme={null}
waf_policies = await gateway_handler.get_application_gateway_waf_policies(resource_group_name="my-rg")
print(waf_policies)
```
**Get Public IPs:**
Retrieve public IP addresses available for Application Gateways.
```python theme={null}
public_ips = await gateway_handler.get_public_ip_addresses(resource_group_name="my-rg")
print(public_ips)
```
**Get Virtual Networks:**
List all VNets and subnets associated with Application Gateways.
```python theme={null}
vnets = await gateway_handler.get_virtual_networks(resource_group_name="my-rg")
print(vnets)
```
**Start Application Gateway:**
Start an Application Gateway instance.
```python theme={null}
await gateway_handler.start_application_gateway(resource_group_name="my-rg", gateway_name="my-gateway")
```
**Stop Application Gateway:**
Stop an Application Gateway instance.
```python theme={null}
await gateway_handler.stop_application_gateway(resource_group_name="my-rg", gateway_name="my-gateway")
```
**Collect All Data:**
Fetch comprehensive data including gateways, WAF policies, backend health, public IPs, and VNets.
```python theme={null}
data = await gateway_handler.collect_all_application_gateway_data(resource_group_name="my-rg")
print(data)
```
**Authentication & Authorization Details:**
Collect authentication, authorization, and RBAC configuration details for API Management linked with gateways.
```python theme={null}
auth_data = await gateway_handler.collect_authentication_authorization_details(resource_group_name="my-rg")
print(auth_data)
```
# Cloud Functions
Source: https://docs.superagentx.ai/examples/handlers/azure/cloud_functions
Azure Functions is a **serverless compute service** that enables you to run event-driven code without provisioning or managing infrastructure.
It allows developers to write functions in multiple languages (like Python, Node.js, C#, etc.) and execute them in response to triggers (HTTP requests, timers, storage events, queues, etc.).
Using the `AzureFunctionHandler`, you can easily **list, retrieve, and configure** function apps within your Azure subscription. It leverages the Azure SDK to interact with Azure Functions resources, supporting authentication via `ClientSecretCredential`.
## Example
To create the `AzureFunctionHandler` object, provide your **Azure credentials**:
```python theme={null}
import os
from superagentx_handlers.azure.functions import AzureFunctionHandler
azure_handler = AzureFunctionHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET"),
)
```
**List All Function Apps in a Subscription:**
Retrieves all function apps across the given subscription.
```python theme={null}
function_apps = await azure_handler.get_all_function_apps_in_subscription()
print(function_apps)
```
**List Function Apps by Resource Group:**
Get all function apps inside a specific resource group.
```python theme={null}
apps_in_rg = await azure_handler.get_function_apps_by_resource_group("my-resource-group")
print(apps_in_rg)
```
**Get a Specific Function App by Name:**
Retrieve metadata and configuration of a specific function app.
```python theme={null}
function_app = await azure_handler.get_function_app_by_name(
resource_group_name="my-resource-group",
function_app_name="my-function-app"
)
print(function_app)
```
**Get Function App Configuration:**
Fetches site configuration, app settings, and connection strings.
```python theme={null}
config = await azure_handler.get_function_app_configuration(
resource_group_name="my-resource-group",
function_app_name="my-function-app"
)
print(config)
```
# Entra IAM Handler
Source: https://docs.superagentx.ai/examples/handlers/azure/iam
Microsoft Entra IAM (Identity and Access Management) is part of Microsoft Entra ID (formerly Azure Active Directory). It helps manage **users, groups, applications, roles, and MFA (multi-factor authentication)**. Using Microsoft Graph APIs, you can collect IAM evidence for governance, compliance, and auditing purposes.
This handler provides methods to retrieve IAM data like users, groups, service principals (applications), role definitions, and MFA status.
## Example
To create the **EntraIAMHandler** object, initialize it with your **Entra Tenant ID, Client ID, and Client Secret** (or load them from environment variables).
```python theme={null}
import os
from superagentx_handlers.azure.iam import EntraIAMHandler
entra_handler = EntraIAMHandler(
tenant_id=os.getenv("ENTRA_TENANT_ID"),
client_id=os.getenv("ENTRA_CLIENT_ID"),
client_secret=os.getenv("ENTRA_CLIENT_SECRET")
)
```
**Collect Users IAM Evidence:**
Fetches all users and their IAM-related details (display name, UPN, email, type, assigned roles).
Requires User.Read.All and optionally RoleManagement.Read.All.
```python theme={null}
users = await entra_handler.collect_users_iam_evidence()
print(users)
```
**Collect Groups IAM Evidence:**
Fetches all groups and their members (users, devices, service principals).
Requires Group.Read.All.
```python theme={null}
groups = await entra_handler.collect_groups_iam_evidence()
print(groups)
```
**Collect Applications IAM Evidence (Service Principals):**
Retrieves all applications (service principals) and their owners.
Requires Application.Read.All.
```python theme={null}
apps = await entra_handler.collect_applications_iam_evidence()
print(apps)
```
**Collect Role Definitions:**
Retrieves all built-in and custom role definitions available in Microsoft Entra ID.
Requires RoleManagement.Read.Directory.
```python theme={null}
roles = await entra_handler.collect_roles_definitions()
print(roles)
```
**Collect MFA Status Evidence:**
Collects MFA registration status and recent MFA usage from sign-in logs for users.
Requires Reports.Read.All, UserAuthenticationMethod.Read.All, and AuditLog.Read.All.
```python theme={null}
mfa = await entra_handler.collect_mfa_status_evidence(days_ago=30)
print(mfa)
```
**Collect All entra IAM Evidence (Summary):**
Fetches users, groups, applications, roles, and MFA evidence in a single call.
```python theme={null}
all_evidence = await entra_handler.collect_all_entra_iam_evidence()
print(all_evidence)
```
# Load Balancer Handler
Source: https://docs.superagentx.ai/examples/handlers/azure/load_balancer
The **Azure Load Balancer Handler** is responsible for retrieving and managing **Azure Load Balancers** within a subscription.
It uses the Azure SDK for Python (`azure-mgmt-network`) and authentication via `azure-identity` to collect server asset details for governance, compliance, and monitoring.
## Example
To create the **AzureLoadBalancerHandler** object, initialize it with your **Azure Subscription ID, Tenant ID, Client ID, and Client Secret** (or load them from environment variables):
```python theme={null}
import os
from superagentx_handlers.azure.loadbalancer import AzureLoadBalancerHandler
lb_handler = AzureLoadBalancerHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET")
)
```
**Get All Load Balancers in Subscription:**
Fetches all load balancers in the configured subscription, including their properties (frontend IPs, backend pools, probes, rules, etc.).
```python theme={null}
lbs = await lb_handler.get_all_load_balancers_in_subscription()
print(lbs)
```
# NSG Handler
Source: https://docs.superagentx.ai/examples/handlers/azure/nsg
Azure Network Security Group (NSG) is a core networking feature in Microsoft Azure that acts as a virtual firewall.
It controls inbound and outbound traffic to Azure resources such as Virtual Machines, Subnets, and NICs.
The `AzureNSGHandler` makes it easy to authenticate with Azure using service principal credentials and retrieve
all NSGs within a subscription, helping with **network security auditing** and **firewall configuration visibility**.
## Example
To create the `AzureNSGHandler` object, provide your Azure credentials (subscription ID, tenant ID, client ID, and client secret).
These can also be loaded from environment variables:
```python theme={null}
import os
from superagentx_handlers.azure.nsg import AzureNSGHandler
nsg_handler = AzureNSGHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET"),
)
```
**List All Network Security Groups (NSGs):**
Retrieves a list of all NSGs in the subscription. Each NSG includes its rules and properties.
```python theme={null}
nsgs = await nsg_handler.get_network_security_groups()
print(nsgs)
```
# Storage Handler
Source: https://docs.superagentx.ai/examples/handlers/azure/storage
Azure Storage Accounts are scalable cloud storage solutions in Microsoft Azure,
commonly used to store **blobs (files), queues, tables, and file shares**.
The `AzureStorageHandler` provides an abstraction layer to authenticate with Azure using
service principal credentials and retrieve all storage accounts within a subscription,
including their **encryption settings, firewall/network rules, and public access status**.
## Example
To create the `AzureStorageHandler` object, provide your Azure credentials
(subscription ID, tenant ID, client ID, and client secret).
These can also be loaded from environment variables:
```python theme={null}
import os
from superagentx_handlers.azure.storage import AzureStorageHandler
storage_handler = AzureStorageHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET"),
)
```
**List All Storage Accounts:**
Retrieves a list of all storage accounts within the subscription,
along with their associated properties.
```python theme={null}
accounts = await storage_handler.list_storage_accounts()
print(accounts)
```
# VM Handler
Source: https://docs.superagentx.ai/examples/handlers/azure/vm
Azure Virtual Machines (VMs) are scalable computing resources in Microsoft Azure.
They allow you to run Windows or Linux servers in the cloud without needing to maintain physical hardware.
With Azure VMs, you can host applications, databases, and workloads with built-in high availability, security, and scalability.
This handler provides seamless interaction with Azure VMs to collect information about virtual machines within a subscription.
It uses service principal authentication (`tenant_id`, `client_id`, `client_secret`) and requires appropriate Azure RBAC permissions (e.g., **Reader** role).
## Example
To create the `AzureVMHandler` object with your Azure credentials:
```python theme={null}
import os
from superagentx_handlers.azure.vm import AzureVMHandler
azure_vm_handler = AzureVMHandler(
subscription_id=os.getenv("AZURE_SUBSCRIPTION_ID"),
tenant_id=os.getenv("AZURE_TENANT_ID"),
client_id=os.getenv("AZURE_CLIENT_ID"),
client_secret=os.getenv("AZURE_CLIENT_SECRET")
)
```
**List Virtual Machines (get\_vms):**
Retrieves a list of all accessible Azure VM instances within the configured subscription.
Each VMโs details (name, ID, region, size, OS, and network/storage profile) are returned in dictionary format.
```python theme={null}
vms = await azure_vm_handler.get_vms()
print(vms)
```
# Content Creator
Source: https://docs.superagentx.ai/examples/handlers/content_creator
These tools enable creators to produce, enhance, and automate content across multiple platforms.
LLMs play a crucial role in streamlining workflows, increasing efficiency, and delivering personalized
experiences to audiences.
## Example
This code sets up a content creation tool using a language model. The configuration specifies the model type and service
(OpenAI). The AIHandler uses the connected LLM (Language Model) client
to generate or process text content efficiently.
To use OpenAI models, you'll need to create an OpenAI account and get an API key. For more details refer here.
> Set up a OpenAI API key as an environmental variable and run the following code.
```python theme={null}
export OPENAI_API_KEY = "**************************"
```
```python content_creator_handler.py theme={null}
import asyncio
from superagentx.handler.ai import AIHandler
from superagentx.llm import LLMClient
async def content_text_create(instruction: str):
llm_config = {
'model': 'gpt-4-turbo-2024-04-09',
'llm_type': 'openai'
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
content_creator_handler = AIHandler(
llm=llm_client
)
result = await content_creator_handler.text_creation(instruction=instruction)
return result.choices[0].message.content
async def main():
instruction = input("Enter your instruction here: ")
res = await content_text_create(instruction=instruction)
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```log content creator theme={null}
Enter Instruction: create robusting content
Out[1]: 'Creating compelling and robust content requires attention to detail,creativity, and strategy.
Whether you\'re creating content for a blog, social media, a website, or any other platform, here are
some key steps to help you produce effective content:\n\n### 1. **Understand Your Audience**\n -
**Research:** Know who your audience is. What are their needs, preferences, and challenges? Use surveys,
social media listening tools, and analytics to gather insights.\n - **Personas:** Develop buyer or reader
personas to better target your content.\n\n### 2. **Define Your Goals**\n - What do you want to achieve with
your content? Goals could include increasing brand awareness, generating leads, educating your audience, or
building community engagement.\n\n### 3. **Choose the Right Format**\n - Depending on your audience and goals,
decide on the best format. Options include blog posts, videos, infographics, podcasts, and more.\n -
Consider multimedia formats to increase engagement and appeal to different learning styles.\n\n###
4. **Craft Compelling Headlines**\n - Your headline is often the first impression. Make it catchy, intriguing,
and clear to draw in readers.\n\n###'
```
# Csv Handler
Source: https://docs.superagentx.ai/examples/handlers/csv_cli
The method leverages a query string provided by the user to generate a filter condition based on the columns of the
CSV file. This condition is used to extract relevant data from the CSV and return it in a structured format. The filtering
logic is generated by an external LLM (Large Language Model) client, which receives a prompt based on the CSV's structure.
## Example
This method is part of the CsvHandler class, which handles CSV file operations, and is specifically designed to search
through the data in the CSV file based on a user-defined query.
To use OpenAI models, youโll need to create an OpenAI account and get an API key. For more details refer here.
> Set up a OpenAI API key as an environmental variable and run the following code.
```python theme={null}
export OPENAI_API_KEY = "**************************"
```
```python csv_handler.py theme={null}
import asyncio
from superagentx_handlers.csv_cli import CsvHandler
from superagentx.llm import LLMClient
async def search(query):
input_path = ""
llm_config = {'llm_type': 'openai'}
llm_client = LLMClient(llm_config=llm_config)
csv_handler = CsvHandler(
file_path=input_path,
llm_client=llm_client
)
return await csv_handler.search(query)
async def main(query):
res = await search(query)
print(res)
if __name__ == '__main__':
query="who are all Petroleum engineer"
asyncio.run(main(query))
```
### Result
```python csv cli theme={null}
{
"Index": {
"59": 60
},
"User Id": {
"59": "0f8deedb629A5f6"
},
"First Name": {
"59": "Grace"
},
"Last Name": {
"59": "Phelps"
},
"Sex": {
"59": "Male"
},
"Email": {
"59": "clarkeangela@example.net"
},
"Phone": {
"59": "(034)867-8827x6777"
},
"Date of birth": {
"59": "1909-10-15"
},
"Job Title": {
"59": "Petroleum engineer"
}
}
```
# Snyk Handler
Source: https://docs.superagentx.ai/examples/handlers/cybersecurity/snyk
Snyk is a developer-first security platform that helps identify and fix vulnerabilities in code, open-source libraries, containers, and infrastructure as code.
It integrates into your development workflow, enabling continuous security monitoring, license compliance checks, and proactive vulnerability management.
The **SnykHandler** provides programmatic access to the Snyk REST API. It allows you to retrieve organizations, projects, issues, vulnerabilities, dependencies, container images, and more. You can also monitor projects and dependencies for vulnerabilities.
## Example
To create the `SnykHandler` object with credentials:
```python theme={null}
import os
from superagentx_handlers.snyk import SnykHandler
snyk_handler = SnykHandler(
api_token=os.getenv("SNYK_API_TOKEN"),
base_url="https://api.snyk.io/rest", # default
version="2024-10-15" # default
)
```
**Get Organizations (get\_organizations):**
Retrieve all organizations accessible to the user.
```python theme={null}
orgs = await snyk_handler.get_organizations()
print(orgs)
```
**Get Projects (get\_projects):**
Retrieve all projects for a specific organization.
```python theme={null}
projects = await snyk_handler.get_projects(org_id="your-org-id")
print(projects)
```
**Get Project Issues (get\_project\_issues):**
Retrieve all issues for a specific project.
```python theme={null}
issues = await snyk_handler.get_project_issues(org_id="your-org-id", project_id="your-project-id")
print(issues)
```
**Get Vulnerabilities (get\_vulnerabilities):**
Retrieve vulnerabilities for an organization, with optional severity filter.
```python theme={null}
vulns = await snyk_handler.get_vulnerabilities(org_id="your-org-id", severity="high")
print(vulns)
```
**Get Licenses (get\_licenses):**
Retrieve license issues for an organization.
```python theme={null}
licenses = await snyk_handler.get_licenses(org_id="your-org-id")
print(licenses)
```
**Get Dependencies (get\_dependencies):**
Retrieve dependencies for a specific project.
```python theme={null}
deps = await snyk_handler.get_dependencies(org_id="your-org-id", project_id="your-project-id")
print(deps)
```
**Get Container Images (get\_container\_images):**
Retrieve container images for an organization.
```python theme={null}
images = await snyk_handler.get_container_images(org_id="your-org-id")
print(images)
```
**Get Container Image Details (get\_container\_image):**
Retrieve details for a specific container image.
```python theme={null}
image = await snyk_handler.get_container_image(org_id="your-org-id", image_id="image-123")
print(image)
```
**Update Container Image Target Refs (update\_container\_image\_target\_refs):**
Update target references for a container image.
```python theme={null}
response = await snyk_handler.update_container_image_target_refs(
org_id="your-org-id",
image_id="image-123",
target_refs_data=[{"type": "target", "id": "target-456"}]
)
print(response)
```
**Monitor Dependency Graph (monitor\_dep\_graph):**
Monitor a dependency graph for vulnerabilities.
```python theme={null}
results = await snyk_handler.monitor_dep_graph(dep_graph_data={"graph": {...}})
print(results)
```
**Test Project (test\_project):**
Test a project for vulnerabilities.
```python theme={null}
results = await snyk_handler.test_project(org_id="your-org-id", project_id="your-project-id")
print(results)
```
**Get Users (get\_users):**
Retrieve users in an organization.
```python theme={null}
users = await snyk_handler.get_users(org_id="your-org-id")
print(users)
```
**Get Integrations (get\_integrations):**
Retrieve integrations for an organization.
```python theme={null}
integrations = await snyk_handler.get_integrations(org_id="your-org-id")
print(integrations)
```
**Get Targets (get\_targets):**
Retrieve targets in an organization.
```python theme={null}
targets = await snyk_handler.get_targets(org_id="your-org-id")
print(targets)
```
**Get Apps (get\_apps):**
Retrieve apps for an organization.
```python theme={null}
apps = await snyk_handler.get_apps(org_id="your-org-id")
print(apps)
```
**Get Vulnerability Details (get\_vulnerability\_details):**
Retrieve detailed information about a specific vulnerability.
```python theme={null}
vuln = await snyk_handler.get_vulnerability_details(vuln_id="CVE-2025-1234")
print(vuln)
```
**Monitor Project (monitor\_project):**
Monitor a project for vulnerabilities.
```python theme={null}
response = await snyk_handler.monitor_project(
org_id="your-org-id",
project_data={"name": "new-project", "source": {...}}
)
print(response)
```
**Collect All Container Data (collect\_all\_container\_data):**
Collect all container-related data for an organization.
```python theme={null}
data = await snyk_handler.collect_all_container_data(org_id="your-org-id")
print(data)
```
**Collect All Snyk Data (collect\_all\_snyk\_data):**
Collect all available Snyk data for an organization (projects, issues, dependencies, vulnerabilities, users, integrations, containers, etc.).
```python theme={null}
data = await snyk_handler.collect_all_snyk_data(org_id="your-org-id")
print(data)
```
# Datadog Incidents
Source: https://docs.superagentx.ai/examples/handlers/datadog/incidents
Datadog Incidents is a service that helps engineering and operations teams track, manage, and resolve incidents efficiently.
It provides powerful integrations, collaborative workflows, and monitoring visibility to ensure quick resolution of critical issues.
This handler allows seamless interaction with **Datadog Incidents API**, enabling you to **list** and **search** incidents programmatically.
It uses API and Application keys for authentication (`DD_API_KEY`, `DD_APP_KEY`) and requires the `DD_SITE` configuration (e.g., `us5.datadoghq.com`).
## Example
To create the `DDIncidentsHandler` object with Datadog credentials:
```python theme={null}
import os
from superagentx_handlers.datadog.incidents import DDIncidentsHandler
dd_handler = DDIncidentsHandler(
host=os.getenv("DD_SITE"), # e.g., "us5.datadoghq.com"
api_key=os.getenv("DD_API_KEY"),
app_key=os.getenv("DD_APP_KEY")
)
```
**List Incidents (list\_incidents):**
Retrieves a list of all incidents in your Datadog account.
```python theme={null}
incidents = await dd_handler.list_incidents()
print(incidents)
```
**Search Incidents (search\_incidents):**
Search for incidents that match a specific query (by title, status, severity, etc.).
```python theme={null}
results = await dd_handler.search_incidents(query="status:active severity:SEV-1")
print(results)
```
# Elastic Search Handler
Source: https://docs.superagentx.ai/examples/handlers/elastic
Elasticsearch is a powerful, open-source search and analytics engine based on Apache Lucene. It quickly stores,
searches, and analyzes large amounts of data by creating indexes, rather than searching the text directly. With
a document-based structure and easy-to-use REST APIs, it processes and returns data in JSON format, offering
fast and scalable performance.
### Setup
Install Elasticsearch locally by running it on your machine. The simplest method is to use the official
Elasticsearch Docker image. For more details, check the Elasticsearch Docker documentation.
## Example
This code initializes an ElasticsearchHandler to connect to an Elasticsearch server running on localhost with specified
username and password. This setup allows for searching and managing data stored in Elasticsearch efficiently.
```python elastic_handler.py theme={null}
import asyncio
from superagentx.handler.elastic_search import ElasticsearchHandler
async def create_index():
elasticsearch_handler = ElasticsearchHandler(
hosts="http://localhost:9200",
username="",
password=""
)
return await elasticsearch_handler.create(
index_name="",
document_id="",
document={
"@timestamp": "2099-11-15T13:12:00",
"message": "GET /search HTTP/1.1 200 1070000",
}
)
async def main():
res = await create_index()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```inline Index Create Output: theme={null}
{
"_index": "project8",
"_id": "project test",
"_version": 1,
"result": "created",
"_shards": {
"total": 2,
"successful": 1,
"failed": 0
},
"_seq_no": 0,
"_primary_term": 1
}
```
# E-mail Handler
Source: https://docs.superagentx.ai/examples/handlers/email
The EmailHandler is a simple and effective tool for managing email communication in your applications.
It allows you to send emails easily, handle multiple recipients, and include attachments, making it
perfect for both personal and professional use.
> **From:** The email address of the sender.
> **To:** The primary recipient(s) of the email.
> **Cc:** Additional recipients who will receive a copy of the email, visible to all.
> **Bcc:** Recipients who receive a copy of the email without other recipients knowing their addresses.
> **Subject:** A brief summary or title of the email's content.
> **Body:** The main content or message of the email, where the sender communicates their thoughts or information.
### Setup
Install Postfix locally by running it on your machine. For more detail refer here.
> **Note:** Postfix can run on UNIX-like systems including AIX, BSD, HP-UX, Linux, MacOS X, Solaris, and more.
## Example
This code initializes an EmailHandler to connect to an email server using the specified host and port number.
This setup enables sending, receiving, or managing emails programmatically.
```python email_handler.py theme={null}
import asyncio
from superagentx.handler.send_email import EmailHandler
async def send_email():
email_handler = EmailHandler(
host="mail.example.com",
port=0
)
body = {
"sender": "sample123@abcd.io",
"to": ["test231@abcdf.io"],
"subject": "test",
"body": "Hi"
}
return await email_handler.send_email(**body)
async def main():
res = await send_email()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
# Exa Search Handler
Source: https://docs.superagentx.ai/examples/handlers/exa
Exa helps you locate precise content online using embeddings-based search. Its API offers three key features:
> **Page Search:** Find web pages through natural language queries, with an option to use traditional keyword searches
> if needed.
**Content Extraction:** Access clean, parsed HTML from search results, with the option to focus on
> relevant sections using the highlights feature.
**Similar Page Discovery:** Input a link to identify and retrieve
> pages with related content or meaning.
## Example
This code sets up an ExaHandler to perform a search using an API key. It retrieves content based on the provided query,
automatically determining the search type and returning up to five results, making it easy to find relevant information
quickly.
To access ExaSearch services, you'll need to create an ExaSearch account, obtain an API key, and subscribe to the
required API for your application. For more details, refer to the
ExaSearch documentation.
> Set up a Exa API key as an environmental variable and run the following code.
```python theme={null}
export EXA_API_KEY = "**************************"
```
```python exa_handler.py theme={null}
import os
import asyncio
from superagentx.handler.exa_search import ExaHandler
async def search_contents(query):
exa_handler = ExaHandler()
return await exa_handler.search_contents(
query=query,
search_type="auto",
use_autoprompt=True,
num_results=5,
)
async def main(query):
res = await search_contents(query)
print(res)
if __name__ == '__main__':
query = "What is Agentic AI Framework?"
asyncio.run(main(query))
```
## Result
```log exa search theme={null}
URL: https://aima.eecs.berkeley.edu/2nd-ed/ai.html
ID: http://aima.eecs.berkeley.edu/2nd-ed/ai.html
Score: 0.19881151616573334
Published Date: 2006-09-12T00:00:00.000Z
Author:
Image: None
Text: This page links to 849 pages around the web with
information on Artificial Intelligence. Links in Bold* followed by a star are especially useful
and interesting sites. Links with a + sign at the end have "tooltip"
information that will pop up if you put your mouse over the link for a
second or two. If you have new links to add, mail them to peter@norvig.com.
```
# AI Handler
Source: https://docs.superagentx.ai/examples/handlers/extract/ai
The **ExtractAIHandler** is a utility for document AI extraction.
It integrates with an AI-powered backend to process files (e.g., invoices, PDFs, receipts) and extract structured JSON data.
It supports Base64 file encoding, asynchronous extraction requests, job polling, and retrieval of processed results.
This handler is useful for **invoice extraction, document digitization, and structured data retrieval** from files.
## Example
To create the `ExtractAIHandler` object with your Extract API credentials:
```python theme={null}
import os
from superagentx.handler.ai.extract import ExtractAIHandler
extract_handler = ExtractAIHandler(
prompt_name="invoice_extraction",
api_token=os.getenv("EXTRACT_API_TOKEN"),
base_url=os.getenv("BASE_URL"),
project_id="test_project_123"
)
```
**Get File Base64 Data:**
Reads a file from the given path and returns its Base64-encoded content.
```python theme={null}
file_data = await extract_handler.get_file_base64_data("invoice.pdf")
print(file_data[:100]) # preview first 100 chars
```
**Extract API:**
Initiates a file extraction request and polls until results are available.
```python theme={null}
result = await extract_handler.extract_api(
file_path="invoice.pdf",
file_data=file_data,
poll_interval=5,
retry=10
)
print(result)
```
**Get Invoice JSON Data:**
Fetches the extracted JSON data using the reference ID of a completed job.
```python theme={null}
invoice_data = await extract_handler.get_invoice_json_data("ref12345")
print(invoice_data)
```
# Financial Data Handler
Source: https://docs.superagentx.ai/examples/handlers/financial_data
The Financial Data Handler is a tool that helps get important financial information from online sources.
It allows businesses or individuals to easily access real-time data like stock prices, company financial
details, and income statements. This information can be used for tracking market trends, analyzing a
companyโs performance, or making better investment decisions.
Refer the link to know more details about symbol from the document
here.
## Example
This code sets up a FinancialHandler to access financial data using an API key and specifies the stock symbol for Amazon
(AMZN). This allows for retrieving stock market information and other financial data programmatically.
To use Financial Modeling, you'll need to create an Financial Modeling account and get an API key. For more details refer here.
> Set up a FINANCIAL\_DATA\_API\_KEY key as an environmental variable and run the following code.
```python theme={null}
export FINANCIAL_DATA_API_KEY = "**************************"
```
```python financial_handler.py theme={null}
import os
import asyncio
from superagentx.handler.financial_data import FinancialHandler
async def get_stock_price():
financial_handler = FinancialHandler(
symbol='AMZN'
)
return await financial_handler.get_stock_price()
async def main():
res = await get_stock_price()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```inline Stock price Output: theme={null}
{
"data": [
{
"symbol": "AA",
"name": "Alcoa Corporation",
"price": 41.71,
"changesPercentage": 3.3705,
"change": 1.36,
"dayLow": 40.86,
"dayHigh": 42.1659,
"yearHigh": 45.48,
"yearLow": 23.07,
"marketCap": 10775361358,
"priceAvg50": 34.709,
"priceAvg200": 34.4335,
"exchange": "NYSE",
"volume": 4489893,
"avgVolume": 6216657,
"open": 41.77,
"previousClose": 40.35,
"eps": -1.56,
"pe": -26.74,
"earningsAnnouncement": "2025-01-15T05:00:00.000+0000",
"sharesOutstanding": 258339999,
"timestamp": 1729281602
}
]
}
```
# GitHub Handler
Source: https://docs.superagentx.ai/examples/handlers/github
GitHub is a leading platform for hosting and collaborating on code repositories. It provides developers with tools for version control, CI/CD workflows, and collaboration across teams.
The **GitHubHandler** allows secure integration with the GitHub API using a personal access token. It enables governance, risk, and compliance (GRC) data collection, such as repository metadata, MFA checks, and other organizational details.
## Example
To create the `GitHubHandler` object with the GitHub credentials:
The `GitHubHandler` connects to the GitHub API using an access token and base URL. It enables seamless interaction with repositories, users, and organizational data.
```python theme={null}
import os
from superagentx_handlers.github.github import GitHubHandler
github_handler = GitHubHandler(
api_base_url=os.getenv("GITHUB_API_BASE_URL", "https://api.github.com"),
github_token=os.getenv("GITHUB_TOKEN")
)
```
**Fetch All Pages:**
The fetch\_all\_pages method retrieves data across paginated GitHub API responses, such as repositories, branches, pull requests, or secrets.
url = f"/user/repos"
headers = github\_handler.\_common\_headers
```python theme={null}
repos = await github_handler.fetch_all_pages(url=url, headers=headers)
print(repos)
```
**Organization Details:**
Fetches details about the authenticated GitHub organization.
```python theme={null}
org = await github_handler.organization_details("my-org")
print(org["login"], org["created_at"])
```
**User Details:**
Retrieves information about the authenticated GitHub user.
```python theme={null}
user = await github_handler.user_details()
print(user["login"], user["email"])
```
**Single MFA Evidence (Internal):**
Checks MFA configuration for a single user (internal helper).
```python theme={null}
mfa_status = await github_handler.single_mfa_evidence_internal("octocat")
print(mfa_status)
```
**MFA Evidence:**
Collects MFA enforcement evidence across organization members.
```python theme={null}
mfa_data = await github_handler.mfa_evidence("my-org")
print(mfa_data)
```
**Process Repositories (Internal):**
Processes repositories internally for compliance checks.
```python theme={null}
repos_data = await github_handler.process_repositories_internal("my-org")
print(len(repos_data))
```
**Repository Summary:**
Summarizes details about a given repository.
```python theme={null}
summary = await github_handler.repository_summary("my-org", "superagentx")
print(summary)
```
**List Organization Members:**
Lists all members in an organization.
```python theme={null}
members = await github_handler.list_organization_members("my-org")
print([m["login"] for m in members])
```
**Fetch Repo Branches (Internal):**
Fetches branch details for a repository (internal helper).
```python theme={null}
branches = await github_handler.fetch_repo_branches_internal("my-org", "superagentx")
print(branches)
```
**Repository Branches:**
Gets all branches of a repository.
```python theme={null}
branches = await github_handler.repository_branches("my-org", "superagentx")
print([b["name"] for b in branches])
```
**Single Branch Protection (Internal):**
Fetches branch protection rules for a branch (internal helper).
```python theme={null}
protection = await github_handler.single_branch_protection_internal("my-org", "superagentx", "main")
print(protection)
```
**Branch Protection:**
Gets branch protection details for all branches in a repository.
```python theme={null}
branch_protection = await github_handler.branch_protection("my-org", "superagentx")
print(branch_protection)
```
**Fetch Pull Requests for Repo (Internal):**
Fetches pull requests for a given repository (internal helper).
```python theme={null}
prs = await github_handler._fetch_pull_requests_for_repo("my-org", "superagentx")
print([pr["title"] for pr in prs])
```
**Pull Requests:**
Gets all pull requests for a repository.
```python theme={null}
prs = await github_handler.pull_requests("my-org", "superagentx")
print(len(prs))
```
**Fetch Dependabot Alerts for Repo (Internal):**
Retrieves Dependabot alerts for a repository (internal helper).
```python theme={null}
alerts = await github_handler._fetch_dependabot_alerts_for_repo("my-org", "superagentx")
print(alerts)
```
**Get Dependabot Alerts:**
Fetches Dependabot alerts across repositories.
```python theme={null}
alerts = await github_handler.get_dependabot_alerts("my-org")
print(alerts)
```
**Get Repository Dependabot Secrets:**
Lists Dependabot secrets for a repository.
```python theme={null}
secrets = await github_handler.get_repository_dependabot_secrets("my-org", "superagentx")
print(secrets)
```
**Fetch Dependencies for Repo (Internal):**
Retrieves dependencies for a repository (internal helper).
```python theme={null}
deps = await github_handler._fetch_dependencies_for_repo("my-org", "superagentx")
print(deps)
```
**List Repository Dependencies:**
Lists dependencies across repositories.
```python theme={null}
dependencies = await github_handler.list_repository_dependencies("my-org", "superagentx")
print(dependencies)
```
**Get Packages:**
Fetches packages for an organization or user.
```python theme={null}
packages = await github_handler.get_packages("my-org")
print([pkg["name"] for pkg in packages])
```
# GitLab Handler
Source: https://docs.superagentx.ai/examples/handlers/gitlab
GitLab is a web-based DevOps platform that provides Git repository management, CI/CD pipelines, issue tracking, and more. It enables teams to collaborate on code, automate software delivery, and enforce security and compliance policies. Using the GitLab API, you can programmatically manage projects, users, groups, pipelines, and repository data for governance and auditing.
## Example
To create the `GitlabHandler` object, initialize it with your GitLab private token and instance URL. The handler authenticates with GitLab and provides methods to fetch projects, issues, merge requests, and more.
```python theme={null}
import os
from superagentx_handlers.gitlab import GitlabHandler
gitlab_handler = GitlabHandler(
private_token=os.getenv("GITLAB_PRIVATE_TOKEN"),
url="https://gitlab.com"
)
```
**Get User Profile:**
Retrieves the full authenticated GitLab user profile.
```python theme={null}
user = await gitlab_handler.get_user_profile()
print(user["username"], user["email"])
```
**Get Projects:**
Lists all projects owned by the authenticated user.
```python theme={null}
projects = await gitlab_handler.get_projects()
print([p["name"] for p in projects])
```
**Get Groups and Members:**
Lists groups owned by the user and their members.
```python theme={null}
groups = await gitlab_handler.get_groups_and_members()
print(groups)
```
**Get Issues:**
Fetches issues for a specific project or across all accessible projects.
```python theme={null}
issues = await gitlab_handler.get_issues(project_id=12345)
print([i["title"] for i in issues])
```
**Get Merge Requests:**
Fetches merge requests for a project or across all projects.
```python theme={null}
mrs = await gitlab_handler.get_merge_requests(project_id=12345)
print([mr["title"] for mr in mrs])
```
**Get Hooks:**
Retrieves webhooks configured for a project (or all projects).
```python theme={null}
hooks = await gitlab_handler.get_hooks(project_id=12345)
print(hooks)
```
**Get Pipelines:**
Lists recent pipelines for a project or across all accessible projects.
```python theme={null}
pipelines = await gitlab_handler.get_pipelines(project_id=12345)
print(pipelines)
```
**Get Branches:**
Fetches all branches for a project or across all projects.
```python theme={null}
branches = await gitlab_handler.get_branches(project_id=12345)
print([b["name"] for b in branches])
```
**Get Branch Protection Rules:**
Fetches branch protection rules for a project (or all projects).
```python theme={null}
rules = await gitlab_handler.get_branch_protection_rules(project_id=12345)
print(rules)
```
**Get Packages:**
Lists all packages published to GitLabโs package registry for a project or across projects.
```python theme={null}
packages = await gitlab_handler.get_packages(project_id=12345)
print([pkg["name"] for pkg in packages])
```
# Hubspot Handler
Source: https://docs.superagentx.ai/examples/handlers/hubspot
This HubSpotHandler class simplifies managing interactions with the HubSpot API, offering seamless methods for
creating, retrieving, and managing contacts, companies, deals, and tickets within a HubSpot environment. By
encapsulating common operations, it ensures that users can efficiently communicate with the API using both
synchronous and asynchronous workflows.
## Example
A HubSpotHandler using an API token from environment variables and creates a new contact in HubSpot with the specified
email, first name, and last name. This allows the application to manage customer information directly within HubSpot.
To use Hubspot Token, you'll need to create an account and get an token. For more details refer
here.
> Set up a HUBSPOT\_TOKEN key as an environmental variable and run the following code.
```python theme={null}
export HUBSPOT_TOKEN = "**************************"
```
```python hubspot_handler.py theme={null}
import os
import asyncio
from superagentx_handlers.crm.hubspot_crm import HubSpotHandler
async def create_contact():
hs_handler = HubSpotHandler()
return await hs_handler.create_contact(
email='ram1234@gmail.com',
first_name="Ram",
last_name="Kumar"
)
async def main():
res = await create_contact()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```python hubspot theme={null}
{'archived': False,
'archived_at': None,
'created_at': datetime.datetime(2024, 10, 22, 5, 46, 6, 926000, tzinfo=tzutc()),
'id': '70443625133',
'properties': {'createdate': '2024-10-22T05:46:06.926Z',
'email': 'ram1234@gmail.com',
'firstname': 'Ram',
'hs_all_contact_vids': '70443625133',
'hs_currently_enrolled_in_prospecting_agent': 'false',
'hs_email_domain': 'gmail.com',
'hs_is_contact': 'true',
'hs_is_unworked': 'true',
'hs_lifecyclestage_lead_date': '2024-10-22T05:46:06.926Z',
'hs_membership_has_accessed_private_content': '0',
'hs_object_id': '70443625133',
'hs_object_source': 'INTEGRATION',
'hs_object_source_id': '4114772',
'hs_object_source_label': 'INTEGRATION',
'hs_pipeline': 'contacts-lifecycle-pipeline',
'hs_registered_member': '0',
'lastmodifieddate': '2024-10-22T05:46:06.926Z',
'lastname': 'Kumar',
'lifecyclestage': 'lead'},
'properties_with_history': None,
'updated_at': datetime.datetime(2024, 10, 22, 5, 46, 6, 926000, tzinfo=tzutc())}
```
# HuggingFace Runner
Source: https://docs.superagentx.ai/examples/handlers/hugging_face
Hugging Face provides a rich ecosystem of state-of-the-art machine learning models for NLP, vision, audio, and multimodal tasks.
Using the transformers pipeline API, models can be loaded and executed with minimal configuration.
The HuggingFaceRunner is an asynchronous machine learning runner designed for agentic and tool-based workflows. It dynamically loads
Hugging Face pipelines, supports CPU and GPU execution, and caches loaded models to avoid redundant initialization.
This makes it efficient, scalable, and ideal for repeated inference workloads.
## Example
To use the HuggingFaceRunner, first initialize the runner and load a pipeline configuration.
The runner automatically caches pipelines based on task, model, device, and data type.
```python theme={null}
from superagentx_handlers.ml.huggingface import HuggingFaceRunner
runner = HuggingFaceRunner()
await runner.load({
"task": "text-classification",
"model_name": "distilbert-base-uncased-finetuned-sst-2-english",
"device": "cpu"
})
```
## Running Inference
**Run Model Inference:**
Executes the loaded pipeline with given inputs and optional parameters.
```python theme={null}
result = await runner.run(
inputs="I really love using this framework!",
params={}
)
print(result)
```
## GPU & Precision Support
**Enable GPU Execution:**
If CUDA is available, the runner automatically switches to GPU execution.
```python theme={null}
await runner.load({
"task": "text-generation",
"model_name": "gpt2",
"device": "gpu"
})
```
**Enable FP16 Precision:**
Reduces memory usage and improves performance on supported GPUs.
```python theme={null}
await runner.load({
"task": "text-generation",
"model_name": "gpt2",
"device": "gpu",
"torch_dtype": "fp16"
})
```
# Universal ML Handler
Source: https://docs.superagentx.ai/examples/handlers/ml_handler
Modern AI systems often rely on multiple machine learning frameworks, each optimized for different tasks. The UniversalMLHandler
provides a unified, framework-agnostic interface for running machine learning inference across different ML ecosystems.
This handler dynamically routes inference requests to the appropriate framework-specific runner (such as Hugging Face, TensorFlow,
or scikit-learn) using an internal router. It abstracts away framework-specific initialization and execution details,
enabling agents and automation workflows to invoke ML models through a single, consistent API.
## Example
To run inference using the UniversalMLHandler, specify the framework, model configuration, and input data.
The handler automatically loads the required runner and executes inference.
```python theme={null}
from superagentx_handlers.ml.universal import UniversalMLHandler
ml_handler = UniversalMLHandler()
```
## Run Inference
**Run Hugging Face Model:**
Executes a Hugging Face pipeline using a unified interface.
```python theme={null}
response = await ml_handler.run(
framework="huggingface",
task="text-classification",
model_name="distilbert-base-uncased-finetuned-sst-2-english",
inputs="This framework is incredibly flexible!",
device="cpu"
)
print(response["output"])
```
# Openapi Handler
Source: https://docs.superagentx.ai/examples/handlers/openapi
The OpenAPIHandler class provides a simple way to interact with APIs that follow the OpenAPI specification.
It allows you to load the OpenAPI specification from either a URL or a local file in JSON or YAML format.
You can easily retrieve all available API endpoints, get the supported HTTP methods for each endpoint,
and view detailed information about specific operations. The class also includes a call\_endpoint method
that makes it easy to send API requests with customizable parameters, request bodies, and headers. This
makes the OpenAPIHandler class a powerful tool for automating interactions with APIs in a Python-based workflow.
```python openapi_handler.py theme={null}
import asyncio
from superagentx.handler.openapi import OpenAPIHandler
async def openapi_handler():
openapi_handler = OpenAPIHandler(
base_url='https://petstore.swagger.io/v2/',
spec_url_path='swagger.json'
)
return await openapi_handler.call_endpoint(
endpoint="/pet/findByStatus",
method="GET",
params={'status': 'sold'})
async def main():
res = await create_contact()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
## Result
```shell theme={null}
platform linux -- Python 3.12.3, pytest-8.3.3, pluggy-1.5.0
rootdir: /home/ram/Projects/superagentX
configfile: pyproject.toml
plugins: asyncio-0.24.0, anyio-4.6.2.post1
asyncio: mode=Mode.AUTO, default_loop_scope=None
collected 1 item
tests/handlers/test_openapi_spec_handler.py::TestOpenAPIHandler::test_openapi_handler
INFO tests.handlers.test_openapi_spec_handler:test_openapi_spec_handler.py:30 Response
[{'id': 360274, 'category': {'id': 126898, 'name': 'tbhBBPKt'}, 'name': 'doggie', 'photoUrls': ['AcPwDp'],
'tags': [{'id': 530594, 'name': 'cgAhLGLz'}], 'status': 'sold'}, {'id': 7, 'category': {'id': 51, 'name': 'cat'},
'name': 'Milo', 'photoUrls': ['string'], 'tags': [{'id': 871, 'name': 'siamese'}], 'status': 'sold'}, {'id': 13,
'category': {'id': 0, 'name': 'cats'}, 'photoUrls': ['string'], 'tags': [{'id': 0, 'name': 'string'}], 'status': 'sold'},
{'id': 102, 'category': {'id': 102, 'name': 'Qwerty'}, 'name': 'Mufasa', 'photoUrls': ['string'], 'tags': [{'id': 102,
'name': 'Mufasa'}], 'status': 'sold'}, {'id': 1346780, 'name': '่ฆ้ฟ่งๅไปฎไป!$%&()*+,-ฦฦ
ฦแธแธแนสถสทสธฤฤฤ
', 'photoUrls':
['https://petstore3.swagger.io/resources/photos/623389095.jpg'], 'tags': [], 'status': 'sold'}]
PASSED [100%]
```
# Scrape Handler
Source: https://docs.superagentx.ai/examples/handlers/scrape
A scrape handler is a tool or part of a program that helps you "scrape" data from websites. When you want to collect
information from a webpage, like product prices, weather data, or news headlines, a scrape handler is the part of the
system that goes to the website, gets the data, and then processes it for you.
```python scrape_handler.py theme={null}
import asyncio
from superagentx_handlers.scrape import ScrapeHandler
async def scrap_content() -> ScrapeHandler:
scrap_handler = ScrapeHandler()
return await scrap_handler.scrap_content(domain_urls=["https://www.booking.com/"])
async def main():
res = await scrap_content()
print(res)
if __name__ == '__main__':
asyncio.run(main())
```
### Result
```shell theme={null}
Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However,
please note that the synchronous version will be deprecated soon.
[LOG] ๐ค๏ธ Warming up the AsyncWebCrawler
[LOG] ๐ AsyncWebCrawler is ready to crawl
[LOG] ๐ธ๏ธ Crawling https://www.booking.com/ using AsyncPlaywrightCrawlerStrategy...
[LOG] โ
Crawled https://www.booking.com/ successfully!
[LOG] ๐ Crawling done for https://www.booking.com/, success: True, time taken: 5.25 seconds
[LOG] ๐ Content extracted for https://www.booking.com/, success: True, time taken: 0.78 seconds
[LOG] ๐ฅ Extracting semantic blocks for https://www.booking.com/, Strategy: AsyncWebCrawler
[LOG] ๐ Extraction done for https://www.booking.com/, time taken: 0.78 seconds.
['[ Skip to main content ](#indexsearch)\n\n[](https://www.booking.com/index.en-gb.html?label=gen173nr-
1BCAEoggI46AdIM1gEaGyIAQGYAQm4ARfIAQzYAQHoAQGIAgGoAgO4At7Q67kGwAIB0gIkMjhlOTViMzAtOWI3Yy00NTM4LWEwNmEtOD
I5NWJjMWFjOWJi2AIF4AIB&sid=e80cd7587c79910cad64ebb4460be87c&aid=304142)\n\nINR\n\n\n\n[](https://secure.booking.com/help.en-gb.html
?label=gen173nr-1BCAEoggI46AdIM1gEaGyIAQGYAQm4ARfIAQzYAQHoAQGIAgGoAgO4At7Q67kGwAIB0gIkMjhlOTViMzAtOWI3Yy00NT
M4LWEwNmEtODI5NWJjMWFjOWJi2AIF4AIB&sid=e80cd7587c79910cad64ebb4460be87c&aid=304142&source=header&src=profile_
contact_cs)[List your property](https://join.booking.com/?label=gen173nr-1BCAEoggI46AdIM1gEaGyIAQGYAQm4ARfIAQz
YAQHoAQGIAgGoAgO4At7Q67kGwAIB0gIkMjhlOTViMzAtOWI3Yy00NTM4LWEwNmEtODI5NWJjMWFjOWJi2AIF4AIB&sid=e80cd7587c79910ca
d64ebb4460be87c&aid=304142&lang=en-gb&utm_medium=frontend&utm_source=topbar)
```
# Serper_dev Handler
Source: https://docs.superagentx.ai/examples/handlers/serper_dev
The Serper.dev API provides a way to conduct real-time web searches, mainly leveraging Google's search capabilities,
and is useful for tasks requiring quick access to up-to-date information. It focuses on returning specific search
results, such as snippets and relevant URLs, making it ideal for developers who need web-based insights integrated
into their applications.
## Example
This code connects to the Serper Dev Tool and searches for the best city to visit based on factors like weather, season,
and prices, returning the top result. It helps users find ideal travel destinations easily.
To use SERPER\_API\_KEY, you'll need to create an account and get an API key. For more details refer here.
> Set up a SERPER\_API\_KEY as an environmental variable and run the following code.
```python theme={null}
export SERPER_API_KEY = "**************************"
```
```python serper_handler.py theme={null}
from superagentx_handlers.serper_dev import SerperDevToolHandler
async def search(query):
serper_dev_handler = SerperDevToolHandler()
return await serper_dev_handler.search(
query=query,
total_results=1
)
async def main(query):
res = await search(query)
print(res)
if __name__ == '__main__':
query='Select the best city based on weather, season, and prices'
asyncio.run(main(query))
```
## Result
```python serper dev theme={null}
[{'attributes': {'Missing': 'Select | Show results with:Select'},
'date': 'Oct 31, 2023',
'link': 'https://www.reddit.com/r/geography/comments/17kklit/what_city_is_the_best_combination_of_weather_and/',
'position': 1,
'sitelinks': [{'link': 'https://www.reddit.com/r/geography/comments/1emkt8r/in_your_opinion_which_us_city_has_the_best/',
'title': 'In your opinion, which U.S. city has the best '
'combination of cost of ...'},
{'link': 'https://www.reddit.com/r/SameGrassButGreener/comments/1drjzbg/what_city_state_has_the_best_yearround_weather/',
'title': 'What city / state has the best year-round weather? '
'Excluding ...'},
{'link': 'https://www.reddit.com/r/SameGrassButGreener/comments/1bn2jh3/what_us_citymetro_area_do_you_feel_has_the_best/',
'title': 'What us city/metro area do you feel has the best '
'weather? - Reddit'},
{'link': 'https://www.reddit.com/r/SameGrassButGreener/comments/1clkiuy/which_cities_or_areas_become_way_more_appealing/',
'title': 'Which cities (or areas) become way more appealing '
'if you can skip ...'}],
'snippet': 'Lisbon has my vote. Enough of a temperature shift to feel '
"seasons, but not so much that it's absurd, especially if you "
'prefer warm to cold.',
'title': 'What city is the BEST combination of weather and cost of living?'}]
```
# ServiceNow Handler
Source: https://docs.superagentx.ai/examples/handlers/servicenow
ServiceNow is a cloud-based platform that provides IT service management (ITSM), helping organizations manage incidents, assets, changes, and workflows. The `ServiceNowHandler` allows seamless integration with ServiceNow to fetch users, assets, and related tickets programmatically.
## Example
To create the `ServiceNowHandler` object, initialize it with your ServiceNow instance URL and credentials. If not provided, the values will be picked up from environment variables.
```python theme={null}
import os
from superagentx_handlers.servicenow import ServiceNowHandler
servicenow_handler = ServiceNowHandler(
instance_url=os.getenv("SERVICENOW_INSTANCE_URL"),
username=os.getenv("SERVICENOW_USERNAME"),
password=os.getenv("SERVICENOW_PASSWORD"),
)
```
**Get User Name:**
Fetches the userโs display name from ServiceNow using their sys\_id.
```python theme={null}
user_name = await servicenow_handler.get_user_name("46d44a2d1b2123009b5f3d92cc4bcb65")
print(user_name) # e.g., "John Doe"
```
**Get Assets with Details and Tickets:**
Retrieves a list of assets along with details like owner, model, warranty, and their associated incident tickets.
```python theme={null}
assets_report = await servicenow_handler.get_assets_with_details_and_tickets()
for asset in assets_report:
print(asset["Asset Name"], asset["Tickets"])
```
# Slack Handler
Source: https://docs.superagentx.ai/examples/handlers/slack
Slack is a messaging platform widely used for team communication and collaboration. It provides channels for organizing conversations, direct messaging, file sharing, and integrates with external tools via its API. Using the Slack API, you can send messages, fetch messages, and manage channels programmatically.
## Example
To create the `SlackHandler` object, initialize it with your Slack **bot token**.
The handler uses the Slack Web API to interact with channels and messages.
```python theme={null}
import os
from superagentx_handlers.slack import SlackHandler
slack_handler = SlackHandler(
bot_token=os.getenv("SLACK_BOT_TOKEN")
)
```
**Send Slack Message:**
Sends a plain text message to a specific Slack channel.
```python theme={null}
response = await slack_handler.send_slack_message(
text="Hello team! :wave:",
channel_id="C1234567890"
)
print(response)
```
**Get Messages from a Channel:**
Retrieves the most recent messages from a given Slack channel.
```python theme={null}
messages = await slack_handler.get_messages_from_channel(
channel_id="C1234567890",
limit=5
)
for msg in messages:
print(msg["text"])
```
**Get Channel ID by Name:**
Looks up the channel ID using its name.
```python theme={null}
channel_id = await slack_handler.get_channel_id("general")
print("Channel ID:", channel_id)
```
# Sql Handler
Source: https://docs.superagentx.ai/examples/handlers/sql
A SQL handler refers to a tool or interface that simplifies the process of interacting with SQL databases.
Its main purpose is to help developers run SQL queries, manage data, and connect to databases with minimal
complexity. It abstracts away repetitive tasks such as writing complex SQL syntax, handling connections,
and managing query results, making it easier to work with databases.
By modifying the database\_type parameter, users can easily switch among various SQL databases while keeping their code
largely unchanged.
> **SQLite** , **MYSQL** , **Postgres** , **Mariadb** , **Oracle** , **Mssql**
## Example
This code connects to a SQLite database (designated as DATABASE) and prepares to insert data into a table named test.
It specifies the values to be added, making it easy to update the database with new information.
```python sql_handler.py theme={null}
from superagentx_handlers.sql import SQLHandler
async def sql_handler():
sql_handler = SQLHandler(
database_type="sqlite",
database=""
)
stmt = "insert into test (x, y) values (:x, :y)"
values = [{
"x": 23,
"y": 24
}]
return await sql_handler.insert(stmt=stmt, values=values)
```
## Result
```python sqlite theme={null}
{
"root_connection": "",
"_dbapi_connection": "",
"dialect": "",
"extracted_parameters": [
{
"name": "x",
"value": null,
"type": "NullType"
},
{
"name": "y",
"value": null,
"type": "NullType"
}
],
"invoked_statement": "",
"compiled": "",
"cache_hit": "CACHE_MISS",
"execution_options": {},
"result_column_struct": {
"columns": [],
"has_rows": true,
"is_complete": false,
"was_updated": false,
"was_deleted": false
},
"is_insert": false,
"is_update": false,
"is_delete": false,
"is_text_statement": true,
"compiled_parameters": [
{
"x": 23,
"y": 24
}
],
"unicode_statement": "insert into test9 (x, y) values (?,?)",
"is_server_side": false,
"cursor": "",
"statement": "insert into test9 (x, y) values (?,?)",
"parameters": [
[23, 24]
],
"row_count": 1,
"is_soft_closed": true
}
}
```
# Taiga Handler
Source: https://docs.superagentx.ai/examples/handlers/taiga
Taiga is an open-source project management platform built for agile teams. It supports epics, user stories,
tasks, and issues with configurable workflows.The TaigaHandler provides an asynchronous interface to the
Taiga REST API, enabling secure authentication and efficient retrieval of project and workflow data for
automation and agent-driven workflows.
## Example
To create a TaigaHandler, initialize it with your Taiga username and password.
The handler authenticates automatically and reuses a single HTTP session for optimal performance.
```python theme={null}
from superagentx_handlers.taiga import TaigaHandler
taiga_handler = TaigaHandler(
username="your_username",
password="your_password"
)
```
**Get Projects:**
Retrieves all projects where the authenticated user is a member.
```python theme={null}
projects = await taiga_handler.get_projects()
for project in projects:
print(project["name"])
```
**Get Workflow Stages:**
Fetches workflow stages for a specific project and entity type.
```python theme={null}
stages = await taiga_handler.get_stages(
kind="task-statuses",
project_id=12345
)
print(stages)
```
**Get All Processes:**
Aggregates all workflow stages (epics, user stories, tasks, and issues) across every project the user belongs to.
```python theme={null}
processes = await taiga_handler.get_all_processes()
for project, workflows in processes.items():
print(project, workflows.keys())
```
**Session Management:**
The handler maintains a reusable asynchronous HTTP session and handles authentication transparently.
To gracefully close the session:
```python theme={null}
await taiga_handler.close()
```
# Twitter Handler
Source: https://docs.superagentx.ai/examples/handlers/twitter
TwitterHandler is an asynchronous tool to post tweets with optional hashtags, user mentions,
and links. Links are automatically shortened using TinyURL. It uses Tweepyโs AsyncClient to
interact with the Twitter API efficiently, ensuring easy and flexible tweet creation.
## Example
To create the TwitterHandler object, initialize it with your Twitter API credentials.
The handler uses Tweepyโs AsyncClient to interact with the Twitter API.
```python theme={null}
import os
from superagentx_handlers.twitter import TwitterHandler
twitter_handler = TwitterHandler(
api_key=os.getenv("CONSUMER_KEY"),
api_secret_key=os.getenv("CONSUMER_SECRET"),
access_token=os.getenv("ACCESS_TOKEN"),
access_token_secret=os.getenv("ACCESS_TOKEN_SECRET")
)
```
**Post a Tweet:**
Posts a tweet with optional hashtags, user mentions, and a shortened link.
```python theme={null}
response = await twitter_handler.post_tweet(
text="Launching our new AI feature!",
link="https://example.com/product",
hash_tags=["AI", "MachineLearning"],
user_tags=["openai"]
)
print(response)
```
# Visualization
Source: https://docs.superagentx.ai/examples/handlers/visualization
A visualization handler is a tool or library that helps developers and analysts create visual representations of data.
Its main purpose is to transform raw data into charts, graphs, and dashboards, making it easier to understand trends,
patterns, and insights. It serves as a bridge between complex datasets and easy-to-digest visuals, which are critical
for decision-making.
## Example
```python visualization.py theme={null}
from superagentx.visualization import Visualize
async def visualize_handler():
obj = Visualize()
chart_data = [
{
"Apples": 5,
"Pears": 3,
"Nectarines": 4,
"Plums": 2
"Grapes": 4,
"Strawberries": 6
},
{
"Apples": 12,
"Pears": 42,
"Nectarines": 1,
"Plums": 51,
"Grapes": 9,
"Strawberries": 21
}
]
await obj.pie_chart(data=chart_data, show_output=True)
```
## Result
![]()
# Weather Handler
Source: https://docs.superagentx.ai/examples/handlers/weather
A weather handler is a tool or library that helps developers integrate weather data into applications.
It works by retrieving real-time or forecasted weather information from external sources, such as weather
APIs, and presenting it in a way that is easy to use in applications like websites, mobile apps, or IoT devices.
## Example
This code initializes a WeatherHandler, which is designed to fetch weather-related data. This setup enables the
application to access various weather services for retrieving current latitude and longitude.
```python weather_handler.py theme={null}
import asyncio
from superagentx_handlers.weather import WeatherHandler
weather_handler = WeatherHandler()
async def get_lat_long(place):
return await weather_handler.get_lat_long(place=place)
async def get_weather(place):
lat_long = await get_lat_long(place)
latitude = lat_long.get("latitude")
longitude = lat_long.get("longitude")
return await weather_handler.get_weather(
latitude=latitude,
longitude=longitude
)
async def main(place):
res = await get_weather(place)
print(res)
if __name__ == '__main__':
place="coimbatore"
asyncio.run(main(place))
```
## Result
```python weather theme={null}
{
"latitude": 11.0,
"longitude": 77.0,
"generationtime_ms": 0.04398822784423828,
"utc_offset_seconds": 0,
"timezone": "GMT",
"timezone_abbreviation": "GMT",
"elevation": 421.0,
"current_weather_units": {
"time": "iso8601",
"interval": "seconds",
"temperature": "ยฐC",
"windspeed": "km/h",
"winddirection": "ยฐ",
"is_day": "",
"weathercode": "wmo code"
},
"current_weather": {
"time": "2024-11-04T17: 00",
"interval": 900,
"temperature": 23.5,
"windspeed": 3.3,
"winddirection": 13,
"is_day": 0,
"weathercode": 80
}
}
```
# WhatsApp Handler
Source: https://docs.superagentx.ai/examples/handlers/whatsapp
WhatsApp is one of the most widely used messaging platforms for real-time communication. The WhatsAppHandler enables agents and
automation workflows to send WhatsApp text messages programmatically using the WhatsApp Cloud API.
This handler abstracts authentication, request formatting, and error handling, allowing seamless integration with WhatsApp for
notifications, alerts, approvals, and conversational workflows.
## Example
## Initialize WhatsAppHandler
Create the handler using explicit parameters or environment variables.
```python theme={null}
from superagentx_handlers.communication.whatsapp import WhatsappHandler
whatsapp_handler = WhatsappHandler(
phone_number_id="123456789012345",
access_token="your_access_token"
)
```
## Send WhatsApp Message
**Send a Text Message:**
Sends a plain text message to a WhatsApp user using their phone number in E.164 format.
```python theme={null}
response = await whatsapp_handler.send_whatsapp_message(
to_number="919876543210",
message_text="๐ Your deployment completed successfully!"
)
print(response)
```
# Wikipedia Handler
Source: https://docs.superagentx.ai/examples/handlers/wikipedia
A Wikipedia handler is a tool or library that simplifies access to Wikipedia's vast knowledge base.
It enables developers to retrieve and manipulate Wikipedia content programmatically, making it useful
for applications requiring summarized, factual, or historical information from the platform. This handler
streamlines interactions with the Wikipedia API, ensuring easy extraction of data such as articles, search
results, and metadata.
## Example
This code initializes a WikipediaHandler to retrieve a summary for a specified topic from Wikipedia. The query is
currently empty, meaning users can input any subject they want to learn about, and the handler will provide a concise overview.
```python wikipedia_handler.py theme={null}
import asyncio
from superagentx_handlers.wikipedia import WikipediaHandler
async def get_summary(query):
search = WikipediaHandler()
return await search.get_summary(
query=query,
)
async def main(query):
res = await get_summary(query)
print(res)
if __name__ == '__main__':
query="Sachin Tendulkar"
asyncio.run(main(query))
```
## Result
```python wikipedia theme={null}
('Sachin Ramesh Tendulkar ( ; pronounced [sษtษin teหษณษulkษษพ]; born 24 April '
'1973) is an Indian former international cricketer who captained the Indian '
'national team. He is widely regarded as one of the greatest batsmen in the '
"history of cricket. Hailed as the world's most prolific batsman of all time, "
'he is the all-time highest run-scorer in both ODI and Test cricket with more '
'than 18,000 runs and 15,000 runs, respectively. He also holds the record for '
'receiving the most player of the match awards in international cricket. '
'Tendulkar was a Member of Parliament, Rajya Sabha by presidential nomination '
'from 2012 to 2018.\n'
'Tendulkar took up cricket at the age of eleven, made his Test match debut on '
'15 November 1989 against Pakistan in Karachi at the age of sixteen, and went '
'on to represent Mumbai domestically and India internationally for over 24 '
'years. In 2002, halfway through his career, Wisden ranked him the '
'second-greatest Test batsman of all time, behind Don Bradman, and the '
'second-greatest ODI batsman of all time, behind Viv Richards. The same year, '
'Tendulkar was a part of the team that was one of the joint-winners of the '
'2002 ICC Champions Trophy. Later in his career, Tendulkar was part of the '
'Indian team that won the 2011 Cricket World Cup, his first win in six World '
'Cup appearances for India. He had previously been named "Player of the '
'Tournament" at the 2003 World Cup.\n'
'Tendulkar has received several awards from the government of India: the '
'Arjuna Award (1994), the Khel Ratna Award (1997), the Padma Shri (1998), and '
'the Padma Vibhushan (2008). After Tendulkar played his last match in '
"November 2013, the Prime Minister's Office announced the decision to award "
"him the Bharat Ratna, India's highest civilian award. He was the first "
'sportsperson to receive the award and, as of 2024, is the youngest '
'recipient. In 2010, Time included Tendulkar in its annual list of the most '
'influential people in the world. Tendulkar was awarded the Sir Garfield '
'Sobers Trophy for cricketer of the year at the 2010 International Cricket '
'Council (ICC) Awards.\n'
'Having retired from ODI cricket in 2012, he retired from all forms of '
'cricket in November 2013 after playing his 200th Test match. Tendulkar '
'played 664 international cricket matches in total, scoring 34,357 runs. In '
'2013, Tendulkar was included in an all-time Test World XI to mark the 150th '
"anniversary of Wisden Cricketers' Almanack, and he was the only specialist "
'batsman of the postโWorld War II era, along with Viv Richards, to get '
'featured in the team. In 2019, he was inducted into the ICC Cricket Hall of '
'Fame.')
```
# Zoho Help Desk Handler
Source: https://docs.superagentx.ai/examples/handlers/zoho_help_desk
Zoho Desk is a cloud-based help desk platform used to manage customer support tickets, workflows, and automation.
The ZohoHelpDeskHandler provides a structured interface for interacting with the Zoho Desk REST API, enabling agent-driven and
automated ticket management.
This handler supports secure OAuth-based authentication and exposes tool-enabled methods to retrieve, update, and transition tickets
across statuses, process stages, and blueprint workflows.
## Example
## Initialize ZohoHelpDeskHandler
Create the handler using explicit parameters.
```python theme={null}
from superagentx_handlers.zoho.helpdesk import ZohoHelpDeskHandler
zoho_handler = ZohoHelpDeskHandler(
org_id="123456789",
auth_token="your_oauth_access_token"
)
```
## Get Ticket
**Retrieve a Ticket by ID:**
Fetches full ticket details including status, fields, and metadata.
```python theme={null}
ticket = zoho_handler.get_ticket("987654321")
print(ticket["subject"], ticket["status"])
```
## Update Ticket
**Update Ticket Fields:**
Allows updating any mutable ticket field such as priority, description, or custom fields.
```python theme={null}
updated_ticket = zoho_handler.update_ticket(
ticket_id="987654321",
fields={
"priority": "High",
"description": "Issue escalated for immediate attention"
}
)
print(updated_ticket)
```
## Change Ticket Status
**Update Ticket Status:**
Transitions a ticket between standard statuses like Open, In Progress, or Closed.
```python theme={null}
zoho_handler.change_status(
ticket_id="987654321",
status="In Progress"
)
```
## Change Process Stage
**Move Ticket to Custom Process Stage:**
Updates a custom field representing internal workflow stages.
```python theme={null}
zoho_handler.change_process_stage(
ticket_id="987654321",
stage_name="L2 Review"
)
```
## Trigger Blueprint Transition
**Execute a Blueprint Transition:**
Triggers a Zoho Blueprint workflow transition using a transition ID.
```python theme={null}
response = zoho_handler.trigger_blueprint_transition(
ticket_id="987654321",
transition_id="456789012"
)
print(response)
)
```
# ZohoHIRS Handler
Source: https://docs.superagentx.ai/examples/handlers/zohohirs
Zoho People (Zoho HIRS) is a cloud-based Human Resource Information System that helps organizations manage employees, onboarding, offboarding, and role changes. Using the Zoho People API, you can programmatically access employee records, workflows, and other HR-related data.
## Example
To create the `ZohoHIRSHandler` object, initialize it with your Zoho **access token**.
The handler provides methods for fetching employee records and HR workflows.
```python theme={null}
import os
from superagentx_handlers.zoho.hirs import ZohoHIRSHandler
zoho_handler = ZohoHIRSHandler(
access_token=os.getenv("ZOHO_PEOPLE_ACCESS_TOKEN")
)
```
**List Employees:**
Retrieves the list of all employees in the Zoho People organization.
```python theme={null}
employees = await zoho_handler.list_employees()
for emp in employees:
print(emp["Name"], emp["Email"])
```
**List Onboarding Workflows:**
Fetches onboarding workflows configured in Zoho People.
```python theme={null}
onboarding_flows = await zoho_handler.list_onboarding_workflows()
print(onboarding_flows)
```
**List Offboarding Workflows:**
Fetches offboarding workflows configured in Zoho People.
```python theme={null}
offboarding_flows = await zoho_handler.list_offboarding_workflows()
print(offboarding_flows)
```
**List Role Change Workflows:**
Retrieves role change (transfer) workflows configured in Zoho People.
```python theme={null}
role_change_flows = await zoho_handler.list_role_change_workflows()
print(role_change_flows)
```
# Handler
Source: https://docs.superagentx.ai/handler
We use the term "handler" interchangeably with "tool calling." While "tool calling" is sometimes used to refer
to the invocation of a single function. It contains the logic for managing particular operations, playing a
key role in organizing and carrying out tasks.
With the `BaseHandler` from `superagentx.handler.base`, you can create your own custom
handlers.
Additionally, we also provide pre-built handlers for various other
services and libraries.
### Tool:
A tool is a method or function that performs a specific action within a larger system. In this context,
tools are asynchronous functions that handle specific tasks, like fetching data from external APIs or
performing calculations. The @tool decorator marks these methods as tools that can be invoked by the system.
```python theme={null}
from superagentx.handler.decorators import tool
class ExaHandler(BaseHandler):
@tool
async def search_contents(
self,
*,
query: str,
use_autoprompt: bool,
num_results: int = 10,
search_type: str | None = None
):
```
# Human Approval
Source: https://docs.superagentx.ai/human_approval
In SuperAgentX, Human-in-the-Loop approval is used as a governance step inside a task or browser agent workflow.
When the agent reaches a sensitive operation (for example: deleting data, triggering payments, or sending external
communications), execution is automatically paused. A human approval request is raised through a configured channel
(UI, notification, or dashboard), and the agent waits for a decision. Once approval is received, the agent resumes
execution; if rejected, the workflow is safely stopped. All approval actions are logged, making the process transparent,
auditable, and fully controlled by humans.
# Installation
Source: https://docs.superagentx.ai/installation
This guide will walk you through the installation process for SuperAgentX and its dependencies. SuperAgentX is a
flexible and powerful AI framework designed to help you efficiently create AI agents and tools. Letโs dive in and
get started!
Ensure that Python version >=3.10 or more is to be installed on your system before continuing.
#### 1. Install SuperAgentX
> Install the main SuperAgentX package using the command given below:
```shell theme={null}
pip install superagentx
```
##### Verify the installation
> To verify that the superagentx is installed correctly, run the following command:
```shell theme={null}
pip show superagentx
```
#### 2. Install SuperAgentX-handlers
> Install the superagentx-handlers using the command given below:
```shell theme={null}
pip install superagentx-handlers
```
#### 3. Install superagentx-examples
> Install the superagentx-examples using the command given below:
```shell theme={null}
pip install superagentx-examples
```
# Introduction
Source: https://docs.superagentx.ai/introduction
**SuperAgentX**: A Lightweight Open Source AI Framework Built for Autonomous Multi-Agent Applications with Artificial General Intelligence (AGI) Capabilities.
## Why SuperAgentX?
**Super** : Advanced AI systems with extraordinary capabilities, pioneering the path to AGI (Artificial General Intelligence) and ASI (Artificial Super Intelligence).
**Agent** : Autonomous Multi AI agent framework designed to make decisions, act independently, and handle complex tasks.
**X** : The unknown, the limitless, the extra factor that makes SuperAgentX revolutionary, futuristic, and transformative.
![]()
***SuperAgentX Features***
๐ **Open-Source Framework**: A lightweight, open-source AI framework built for multi-agent applications with Artificial General Intelligence (AGI) capabilities.
๐๏ธ **Easy Deployment**: Offers WebSocket, RESTful API, IO console and Voice interfaces for rapid setup of agent-based AI solutions.
โจ๏ธ **Streamlined Architecture**: No major dependencies; built independently.
๐ **Contextual Memory**: Uses SQL + Vector databases to store and retrieve user-specific context effectively.
๐ง **Flexible LLM Configuration**: Supports simple configuration options of various Gen AI models.
๐ค๐ป **Extendable Handlers**: Allows integration with diverse APIs, databases, data warehouses, data lakes, IoT streams, and more, making them accessible for function-calling features.
๐ฏ **Goal-Oriented Agents**: Enables the creation of agents with retry mechanisms to achieve set goals.
It provides a powerful, modular, and flexible platform for building autonomous AI agents capable of executing complex tasks with minimal human intervention.
By integrating cutting-edge AI technologies and promoting efficient, scalable agent behavior, SuperAgentX embodies a critical step forward in the path towards superintelligence and AGI.
Whether for research, development, or deployment, SuperAgentX is built to push the boundaries of what is possible with autonomous AI systems.
SuperAgentX installation procedures
Create Your Awesome Multi-Agent Use cases with SuperAgentX
# IO Console
Source: https://docs.superagentx.ai/io-console
This documentation provides an overview of the code used to implement a console-based client for interacting
with the trip planner pipeline using SuperagentX. The client allows users to input search queries via the command
line interface (CLI) and receive results in real-time.
## Overview
The code sets up an interactive CLI using the IOPipe class from SuperagentX. The client allows users to enter a
search query, which is processed by the trip planner pipeline, and results are returned dynamically through the console interface.
## Setup Environment
```shell theme={null}
$ pip install superagentx
```
Implementation
Import the dependencies.
```python theme={null}
import asyncio
from rich import print as rprint
from superagentx.pipeimpl.iopipe import IOPipe
from trip_planner.pipe import get_trip_planner_pipe
```
The main function initializes the trip planner pipeline by calling the get\_trip\_planner\_pipe function.
It then creates an IOPipe object, which serves as the console interface for interacting with the pipeline.
The read\_prompt parameter customizes the prompt displayed to users when they are asked to input a query.
```python theme={null}
async def main():
"""
Launches the trip_planner pipeline console client for processing requests and handling data.
"""
pipe = await get_trip_planner_pipe()
# Create IO Cli Console - Interface
io_pipe = IOPipe(
search_name='SuperAgentX trip_planner',
agentx_pipe=pipe,
read_prompt=f"\n[bold green]Enter your search here"
)
await io_pipe.start()
if __name__ == '__main__':
try:
asyncio.run(main())
except (KeyboardInterrupt, asyncio.CancelledError):
rprint("\nUser canceled the [bold yellow][i]pipe[/i]!")
```
This IO console client provides a simple way for users to interact with the trip planner pipeline, entering search
queries and receiving immediate feedback through the command line interface.
## Run the script
```shell theme={null}
$ python3 iopipe.py
```
## Result
```shell theme={null}
(venv) โ superagentx_examples git:(dev) โ python3.12 trip_planner/trip_planner/iopipe.py
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ SuperAgentX trip_planner โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
Enter your search here: give me a trip plan for goa
INFO:superagentx.agentxpipe:Pipe Trip Planner Pipe starting...
(โ ) Searching...
( โ ) Searching...
(โ ) Searching...
WARNING:chromadb.segment.impl.vector.local_persistent_hnsw:Number of requested results 10 is greater than number of elements in index 1, updating n_results = 1
[LOG] ๐ Crawl4AI 0.3.731
[LOG] ๐ค๏ธ Warming up the AsyncWebCrawler
[LOG] ๐ AsyncWebCrawler is ready to crawl
[LOG] ๐ธ๏ธ Crawling https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0 using AsyncPlaywrightCrawlerStrategy...
[LOG] ๐ธ๏ธ Crawling https://www.nomadasaurus.com/first-timers-guide-goa-india/ using AsyncPlaywrightCrawlerStrategy...
[LOG] ๐ธ๏ธ Crawling https://the-shooting-star.com/perfect-3-day-goa-itinerary/ using AsyncPlaywrightCrawlerStrategy...
[LOG] ๐ธ๏ธ Crawling https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html using AsyncPlaywrightCrawlerStrategy...
[LOG] โ
Crawled https://the-shooting-star.com/perfect-3-day-goa-itinerary/ successfully!
[LOG] ๐ Crawling done for https://the-shooting-star.com/perfect-3-day-goa-itinerary/, success: True, time taken: 2.83 seconds
[LOG] ๐ Content extracted for https://the-shooting-star.com/perfect-3-day-goa-itinerary/, success: True, time taken: 0.90 seconds
[LOG] ๐ฅ Extracting semantic blocks for https://the-shooting-star.com/perfect-3-day-goa-itinerary/, Strategy: AsyncWebCrawler
[LOG] ๐ Extraction done for https://the-shooting-star.com/perfect-3-day-goa-itinerary/, time taken: 0.91 seconds.
[LOG] โ
Crawled https://www.nomadasaurus.com/first-timers-guide-goa-india/ successfully!
[LOG] ๐ Crawling done for https://www.nomadasaurus.com/first-timers-guide-goa-india/, success: True, time taken: 6.08 seconds
[LOG] ๐ Content extracted for https://www.nomadasaurus.com/first-timers-guide-goa-india/, success: True, time taken: 2.28 seconds
[LOG] ๐ฅ Extracting semantic blocks for https://www.nomadasaurus.com/first-timers-guide-goa-india/, Strategy: AsyncWebCrawler
[LOG] ๐ Extraction done for https://www.nomadasaurus.com/first-timers-guide-goa-india/, time taken: 2.29 seconds.
[LOG] โ
Crawled https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0 successfully!
[LOG] ๐ Crawling done for https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0, success: True, time taken: 8.86 seconds
[LOG] ๐ Content extracted for https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0, success: True, time taken: 0.68 seconds
[LOG] ๐ฅ Extracting semantic blocks for https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0, Strategy: AsyncWebCrawler
[LOG] ๐ Extraction done for https://www.tripoto.com/goa/trips/top-things-to-do-in-goa-in-4-days-58a3ef7c589e0, time taken: 0.69 seconds.
[LOG] โ
Crawled https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html successfully!
[LOG] ๐ Crawling done for https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html, success: True, time taken: 9.57 seconds
[LOG] ๐ง Error: After processing the crawled HTML and removing irrelevant tags, nothing was left in the page. Check the markdown for further details.
[LOG] ๐ Content extracted for https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html, success: True, time taken: 0.01 seconds
[LOG] ๐ฅ Extracting semantic blocks for https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html, Strategy: AsyncWebCrawler
[LOG] ๐ Extraction done for https://www.tripadvisor.com/Attractions-g297604-Activities-Goa.html, time taken: 0.01 seconds.
( โ ) Searching...
```
# LLMClient
Source: https://docs.superagentx.ai/llm
### LLM Client object configuration
The `LLMClient` object from `superagentx.llm` is a configuration-driven interface to interact with large language models (LLMs) from various
providers, such as OpenAI, AWSBedrock and more. It allows users to set parameters specific to the model and
provider they intend to use, making it easier to configure different types of LLMs and interact with it.
### Supported LLMs
| Icon | LLM Name |
| :---------------------------------------------------------: | :---------------- |
| ![]( "deepseek")
| **DeepSeek** |
| ![]( "OpenAI")
| **OpenAI** |
| ![]( "Azure OpenAI")
| **Azure OpenAI** |
| ![]( "AWS Bedrock")
| **AWS Bedrock** |
| ![]( "Google Gemini")
| **Google Gemini** |
| ![]( "Meta LLama")
| **Meta Llama** |
| ![]( "Ollama")
| **Ollama** |
| ![]( "Claude AI")
| **Claude AI** |
| ![]( "Mistral AI")
| **Mistral AI** |
| ![]( "IBM WatsonX AI")
| **IBM WatsonX** |
| ![]( "Lite LLM")
| **Lite LLM** |
### Parameters
| Attribute | Parameters | Type | Description |
| :------------------------------- | :------------ | :----- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **LLM Type** | `llm_type` | `str` | This key identifies the service provider or platform from which the language model is sourced. Common values are: โopenaiโ: Refers to OpenAIโs GPT models. |
| **Model** *(optional)* | `model` | `str` | This specifies the language model you wish to use. The model is typically identified by a name or version, such as โgpt-4oโ, โgpt-3.5-turboโ, or another model version that the service provider supports. Default `None` |
| **API Key** *(optional)* | `api_key` | `str` | You can provide the model API key either as a parameter or retrieve it from the environment. Default `None` |
| **Base Url** *(optional)* | `base_url` | `str` | The base URL for the model API. Default `None` |
| **API Version** *(optional)* | `api_version` | `str` | The required API version for the Azure OpenAI llm\_type. Default `None` |
| **Async Mode** *(optional)* | `async_mode` | `bool` | Asynchronous mode of OpenAI or Azure OpenAI client. Default `None` |
| **Embedding Model** *(optional)* | `embed_model` | `str` | Embedding model name, supported models openai, azure-openai, mistral, llama 3.1. Default `None` |
```python DeepSeek theme={null}
from superagentx.llm import LLMClient
llm_config = {
'model': "deepseek-chat",
'llm_type': 'deepseek',
'base_url': 'https://api.deepseek.com/'
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
```
```python OpenAI theme={null}
from superagentx.llm import LLMClient
llm_config = {
"model":'gpt-4o',
"llm_type":'openai'
}
llm_client = LLMClient(llm_config=llm_config)
```
```python AWSBedrock theme={null}
from superagentx.llm import LLMClient
llm_config = {
"model":'anthropic.claude-3-5-sonnet-20241022-v2:0',
"llm_type":'bedrock'
}
llm_client = LLMClient(llm_config=llm_config)
```
```python ollama theme={null}
from superagentx.llm import LLMClient
llm_config = {
'model': 'mistral',
'llm_type': 'ollama',
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
```
```python Claude AI theme={null}
from superagentx.llm import LLMClient
llm_config = {
'model': 'claude-3-7-sonnet-20250219',
'llm_type': 'anthropic'
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
```
```python Lite LLM theme={null}
from superagentx.llm import LLMClient
llm_config = {
'model': 'openai/gpt-5-mini',
'llm_type': 'lite_llm'
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
```
```python Lite LLM theme={null}
from superagentx.llm import LLMClient
llm_config = {
'model': 'anthropic/claude-sonnet-4-5-20250929',
'llm_type': 'lite_llm'
}
llm_client: LLMClient = LLMClient(llm_config=llm_config)
```
### LLM Type and its Values
| Type | value | Example |
| :--------------- | :------------- | :------------------------------------------ |
| **DeepSeek** | `deepseek` | `llm_config = { 'llm_type':'deepseek'}` |
| **Open AI** | `openai` | `llm_config = { 'llm_type':'openai'}` |
| **Azure OpenAI** | `azure-openai` | `llm_config = { 'llm_type':'azure-openai'}` |
| **AWS Bedrock** | `bedrock` | `llm_config = { 'llm_type':'bedrock'}` |
| **Ollama** | `ollama` | `llm_config = { 'llm_type':'ollama'}` |
| **LLama** | `llama` | `llm_config = { 'llm_type':'llama'}` |
| **Gemini** | `gemini` | `llm_config = { 'llm_type':'gemini'}` |
| **Anthropic** | `anthropic` | `llm_config = { 'llm_type':'anthropic'}` |
| **Mistral** | `mistral` | `Coming Soon` |
| **Groq** | `groq` | `Coming Soon` |
| **Together** | `together` | `Coming Soon` |
### OpenAI Supported Models
| Type | Example |
| :------------------------- | :---------------------------------------------------------------------- |
| **gpt-4o** | `llm_config = { "model":'gpt-4o', "llm_type":'openai'}` |
| **gpt-4.1** | `llm_config = { "model":'gpt-4.1', "llm_type":'openai'}` |
| **gpt-4** | `llm_config = { "model":'gpt-4', "llm_type":'openai'}` |
| **gpt-4o-2024-05-13** | `llm_config = { "model":'gpt-4o-2024-05-13', "llm_type":'openai'}` |
| **gpt-4o-2024-08-06** | `llm_config = { "model":'gpt-4o-2024-08-06', "llm_type":'openai'}` |
| **gpt-4-turbo-2024-04-09** | `llm_config = { "model":'gpt-4-turbo-2024-04-09', "llm_type":'openai'}` |
| **gpt-4o-mini-2024-07-18** | `llm_config = { "model":'gpt-4o-mini-2024-07-18', "llm_type":'openai'}` |
| **gpt-4o-mini** | `llm_config = { "model":'gpt-4o-mini', "llm_type":'openai'}` |
### Bedrock Supported Models
| Type | Example |
| :-------------------------------------------- | :------------------------------------------------------------------------------------------ |
| **anthropic.claude-3-5-sonnet-20241022-v2:0** | `llm_config = { "model":'anthropic.claude-3-5-sonnet-20241022-v2:0', "llm_type":'bedrock'}` |
| **anthropic.claude-3-5-haiku-20241022-v1:0** | `llm_config = { "model":'anthropic.claude-3-5-haiku-20241022-v1:0', "llm_type":'bedrock'}` |
| **anthropic.claude-instant-v1:2:100k** | `llm_config = {"model":'anthropic.claude-instant-v1:2:100k',"llm_type":'bedrock'}` |
| **anthropic.claude-3-sonnet-20240229-v1:0** | `llm_config = {"model":'anthropic.claude-3-sonnet-20240229-v1:0',"llm_type":'bedrock'}` |
| **anthropic.claude-3-haiku-20240307-v1:0** | `llm_config = {"model":'anthropic.claude-3-haiku-20240307-v1:0',"llm_type":'bedrock'}` |
| **anthropic.claude-3-opus-20240229-v1:0** | `llm_config = {"model":'anthropic.claude-3-opus-20240229-v1:0',"llm_type":'bedrock'}` |
| **anthropic.claude-3-5-sonnet-20240620-v1:0** | `llm_config = {"model":'anthropic.claude-3-5-sonnet-20240620-v1:0',"llm_type":'bedrock'}` |
| **cohere.command-r-v1:0** | `llm_config = {"model":'cohere.command-r-v1:0',"llm_type":'bedrock'}` |
| **cohere.command-r-plus-v1:0** | `llm_config = {"model":'cohere.command-r-plus-v1:0',"llm_type":'bedrock'}` |
| **meta.llama3-1-8b-instruct-v1:0** | `llm_config = {"model":'meta.llama3-1-8b-instruct-v1:0',"llm_type":'bedrock'}` |
| **meta.llama3-1-70b-instruct-v1:0** | `llm_config = {"model":'meta.llama3-1-70b-instruct-v1:0',"llm_type":'bedrock'}` |
| **meta.llama3-1-405b-instruct-v1:0** | `llm_config = {"model":'meta.llama3-1-405b-instruct-v1:0',"llm_type":'bedrock'}` |
| **mistral.mistral-large-2402-v1:0** | `llm_config = {"model":'mistral.mistral-large-2402-v1:0',"llm_type":'bedrock'}` |
| **mistral.mistral-large-2407-v1:0** | `llm_config = {"model":'mistral.mistral-large-2407-v1:0',"llm_type":'bedrock'}` |
### Ollama Supported Models:
| Type | Example |
| :------------------ | :------------------------------------------------------------- |
| **mistral:latest** | `llm_config = {"model":'mistral:latest',"llm_type":'ollama'}` |
| **llama3.3:latest** | `llm_config = {"model":'llama3.3:latest',"llm_type":'ollama'}` |
# MCP
Source: https://docs.superagentx.ai/mcp
## Introduction:
The Model Context Protocol (MCP) is a lightweight, open protocol designed to standardize how AI applications and
large language models (LLMs) interface with external tools, data sources, and services. It allows developers to
seamlessly integrate LLMs into broader software systems by defining a clear, extensible, and secure context interface.
Think of MCP as a universal adapter for AI systems, enabling smooth interaction between models and the tools they
need โ like local files, databases, or web services.
### Parameters:
| Attribute | Parameters | Description |
| :------------------------ | :--------- | :--------------------------------------------------------------------------------- |
| **Command** *(optional)* | `command` | The command used to launch the MCP-compatible server. Default to `None` |
| **MCP Args** *(optional)* | `mcp_args` | Optional list of additional CLI arguments to pass to the server. Default to `None` |
| **SSE Url** *(optional)* | `sse_url` | Optional Server-Sent Events (SSE) URL for streaming. Default to `None` |
| **Headers** *(optional)* | `headers` | Optional HTTP headers to include in the request. Default to `None` |
| **ENV** *(optional)* | `env` | Optional environment variables to set when running the server. Default to `None` |
### MCP Handler - STDIO:
The STDIO handler communicates using the operating system's standard input and output streams. This method is
typically used when integrating with command-line tools or subprocesses. The MCP (Modular Communication Protocol)
engine sends input to the handler through standard input (stdin) and reads the response from standard output (stdout).
```python theme={null}
from superagentx.handlers.mcp import MCPHandler
mcp_handler = MCPHandler(
command="python",
mcp_args=["-m","mcp_server_reddit"]
)
```
### MCP Handler - SSE:
The SSE handler uses Server-Sent Events, a web-based communication protocol that allows the server to push real-time
updates to the client over a single HTTP connection. The handler remains connected and continuously sends events as
they become available, without the client having to request them each time.
```python theme={null}
from superagentx.handlers.mcp import MCPHandler
mcp_handler = MCPHandler(
sse_url="http://0.0.0.0:8080/sse"
)
```
# Memory
Source: https://docs.superagentx.ai/memory
The `Memory` class from `superagentx.memory` is designed to provide a persistent storage mechanism for storing and retrieving context,
data, and state during interactions with agents in a pipeline. `Memory` allows the system to โrememberโ information
across agent actions and queries, which is particularly useful for complex workflows or tasks that span multiple steps.
The `Memory` class is initialized with a configuration that defines how it interacts with the agent pipeline.
It can be used to persist important data, store intermediate results, or maintain context across different
stages of the agentโs execution.
### Memory config Parameters
| Attribute | Parameters | Description |
| :---------------------------- | :------------- | :---------------------------------------------------------- |
| **LLM Client** | `llm_client` | Configuration for the LLM. |
| **Vector Store** *(optional)* | `vector_store` | Configuration for the vector store. Default is `Chroma db`. |
| **DB Path** *(optional)* | `db_path` | Path to the history database. Default `Sqlite memory`. |
### Parameters
| Attribute | Parameters | Description |
| :---------------- | :-------------- | :------------------- |
| **Memory Config** | `memory_config` | Memory configuration |
```python theme={null}
from superagentx.memory import Memory
memory = Memory(memory_config={"llm_client": llm_client})
```
# Pipe
Source: https://docs.superagentx.ai/pipe
### AgentXPipe
The `AgentXPipe` class from the `superagentx.agentxpipe` module is used to set up a pipeline that allows multiple agents to work
together in sequence, Parallel or Hybrid. In this case, the `AgentXPipe` object enables the use of one or
more agents ***(e.g., ecom\_agent, ai\_agent, content\_creator\_agent etc..)*** in combination with a memory store,
allowing the system to persist context, share information between agents, and maintain state across multiple interactions.
### AgentXPipe Parameters
| Attribute | Parameters | Description |
| :--------------------------------------------- | :--------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **Pipe Id** *(optional)* | `pipe_id` | A unique identifier for the agentxpipe. If not provided, a new UUID will be generated by default. Useful for tracking or referencing the agentxpipe in multi-engine environments. |
| **Name** *(optional)* | `name` | An optional name for the agentxpipe, providing a more friendly reference for display or logging purposes. |
| **Description** *(optional)* | `description` | An optional description that provides additional context or details about the agentxpipe's purpose and capabilities. |
| **Agents** *(optional)* | `agents` | A list of Agent instances (or lists of Agent instances) that are part of this structure. These agents can perform tasks and contribute to achieving the defined goal. |
| **Memory** *(optional)* | `memory` | An optional memory instance that allows the engine to retain information across interactions.This can enhance the pipe's contextual awareness and improve its performance over time. |
| **Stop if goal is not satisfied** *(optional)* | `stop_if_goal_not_satisfied` | A flag indicating whether to stop processing if the goal is not satisfied. When set to True, the agentxpipe operation will halt if the defined goal is not met,preventing any further actions. Defaults to `False`, allowing the process to continue regardless of goal satisfaction. |
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
pipe_id='weather-id-1',
name='Weather Agent',
description="""Weather refers to the atmospheric conditions in
a specific location over a short period of time,
including elements such as temperature, humidity,
precipitation, wind speed, and atmospheric pressure.
It can change rapidly and influences daily activities
and the environment.""",
agents=[lat_long_agent, weather_agent],
memory=None,
stop_if_goal_not_satisfied=False
)
```
## AgentXPipe Agent Configuration
### Sequence
In sequence execution, `agent(s)` are performed one after another, with each `agent(s)` waiting for the
previous one to finish before starting. This approach is simple but can be slow if `agent(s)` are independent.
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...
agents=[agent_1, agent_2, agent_3 ... agent_n],
...
)
# Agents execution will be in following order
# [agent_1, agent_2, agent_3 ... agent_n]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...,
...
)
await pipe.add(agent_1, agent_2)
.
.
.
await pipe.add(agent_3, agent_4, execute_type='SEQUENCE')
.
.
# Agents execution will be in following order
# [agent_1, agent_2, agent_3, agent_4]
```
### Parallel
In parallel execution, `agent(s)` run at the same time, speeding up the process by utilizing multiple threads or processes.
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...
agents=[[agent_1, agent_2, agent_3 ... agent_n]],
...
)
# Agents execution will be in following order
# [[agent_1, agent_2, agent_3 ... agent_n]]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...,
...
)
await pipe.add(agent_1, agent_2, execute_type='PARALLEL')
.
.
.
await pipe.add(agent_3, agent_4, execute_type='PARALLEL')
.
.
# Agents execution will be in following order
# [[agent_1, agent_2, agent_3, agent_4]]
```
### Mixed with Sequence and Parallel
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...
agents=[agent_1, [agent_2, agent_3], agent_4 ... agent_n],
...
)
# Agents execution will be in following order
# [agent_1, [agent_2, agent_3], agent_4, ... agent_n]]
```
#### Alternate way of configure
```python theme={null}
from superagentx.agentxpipe import AgentXPipe
pipe = AgentXPipe(
...,
...
)
await pipe.add(agent_1, agent_2, execute_type='SEQUENCE')
.
.
.
await pipe.add(agent_3, agent_4, execute_type='PARALLEL')
.
.
# Agents execution will be in following order
# [agent_1, agent_2, [agent_3, agent_4]]
```
# PromptTemplate
Source: https://docs.superagentx.ai/prompt
A `PromptTemplate` class is used to create and organize the messages (prompts) that are sent to a language model.
It allows you to define a basic structure for the prompt, making it easy to manage and reuse across different situations.
This helps maintain consistency and simplifies prompt formatting when interacting with the model.
### Parameters
| Attribute | Parameters | Type | Description |
| :------------------------------- | :--------------- | :---- | :------------------------------------------------------------------------------------ |
| **Prompt Type** *(optional)* | `prompt_type` | `str` | specifies the type of prompt. Default `None` |
| **System Message** *(optional)* | `system_message` | `str` | An optional string that defines a system-level message for the prompt. Default `None` |
```python theme={null}
from superagentx.prompt import PromptTemplate
prompt_template = PromptTemplate()
```
# Quickstart
Source: https://docs.superagentx.ai/quickstart
Start building awesome documentation in under 5 minutes
## Introduction
This simple walkthrough will help you get started with building powerful applications using SuperAgentX in just a few steps.
> **1. Installation:** How to install SuperAgentX on your system.
> **2. Create an Application:** Set up a new SuperAgentX project with the basic structure.
> **3. Virtual Environment Setup:** How to create and configure a Python virtual environment for your project.
> **4. Building a Use Case:** Customizing the pipe.py file to fit your application's needs.
> **5. Run the Application:** Finally, how to execute your new project with a simple command.
Note:
After installation, the `superagentx-cli` will be available in the path `~/.local/bin` on `linux`. You may need to add
this to your environment's path variable.
## Create an Application
### For Linux / macOS
```shell theme={null}
$ mkdir <"folderName">
$ python -m venv .venv
$ Source .venv/bin/activate
$ pip install superagentx
```
### For Windows (PowerShell)
```shell theme={null}
$ mkdir <"folderName">
$ python -m venv .venv
$ .venv\Scripts\activate
$ pip install superagentx
```
## Simple Tutorial
```python theme={null}
import asyncio
from superagentx.agent import Agent
from superagentx.engine import Engine
from superagentx.handler.mcp import MCPHandler
from superagentx.llm import LLMClient
from superagentx.prompt import PromptTemplate
async def main():
print("Welcome to SuperAgentX MCP Tutorial")
# Step 1: Initialize LLM
# Note: You need to setup your OpenAI API key before running this step.
# export OPENAI_API_KEY=
llm_config = {"model": "gpt-4o", "llm_type": "openai"}
llm_client = LLMClient(llm_config=llm_config)
# Step 2: Setup MCP tool handler (Reddit trending analyzer)
# Note: You need to install the mcp-server-reddit package before running this step.
# pip install mcp-server-reddit
mcp_handler = MCPHandler(command="python", mcp_args=["-m", "mcp_server_reddit"])
# Step 3: Create Prompt Template
prompt_template = PromptTemplate()
# Step 4: Create Engine for Reddit analysis using MCP
reddit_engine = Engine(handler=mcp_handler, llm=llm_client,
prompt_template=prompt_template)
# Step 5: Define Reddit Agent
agent = Agent(goal="Summarize Reddit Messages",
role="You're Reddit Trend Analyzer",
llm=llm_client, prompt_template=prompt_template,
engines=[reddit_engine])
result = await agent.execute(query_instruction="List top AI Trends")
print(f"Reddit Agent Result : {result}")
if __name__ == "__main__":
asyncio.run(main())
```
## How to Create an Application Using `Cli`
To create a new `superagentx` project, run the following command in your terminal. This will prompt you to create
project with the basic structure set up for your `superagentx`.
```shell theme={null}
$ python3 -m pip install superagentx
```
```shell theme={null}
$ superagentx-cli create
```
### Follow the steps
```shell theme={null}
# First you have to give an application name or folder name.
Enter application name: content creator
# Then you have to enter pipe name or it will take a default application name.
Enter pipe name. Default is application name []: content
# Next you have to give any one of option given below, or it be take a default value all.
Enter one of the option (all, console, websocket, rest) [all]: all
```
After your inputs the `superagentx-cli` create a application like a below structure.
```shell theme={null}
App Name โ๏ธ content creator
Pacakge Name ๐ฆ content_creator
Pipe Name ๐ข content
App Type ๐ ๏ธ all
Creating app at /home/ben/content creator
Creating toml file at /home/ben/content creator/pyproject.toml
Creating readme file at /home/ben/content creator/README.md
Creating pipe file at /home/ben/content creator/content_creator/pipe.py
Your app type selection contains `websocket`, `rest api` option(s).
Enter auth token for `websocket`, `rest api` [5499161e37cb46809791596c0ce49483]:
Creating config file at /home/ben/content creator/content_creator/config.py
Creating iopipe file at /home/ben/content creator/content_creator/iopipe.py
Creating wspipe file at /home/ben/content creator/content_creator/wspipe.py
Creating restpipe file at /home/ben/content creator/content_creator/restpipe.py
```
### Virtual Environment Setup
Now, you'll need to create a Python virtual environment for your project:
```shell theme={null}
$ python3 -m pip install poetry
$ cd /home/ben/Content Creator
$ python3 -m venv venv
$ source venv/bin/activate
(venv) $ poetry install
```
### Build Your Use Case
If you're building a new use case, You need to modify the file at `/home/ben/Content Creator/content_creator/pipe.py`.
The path in this file will be generated based on the input you provide through the SuperAgentX application. In that
file you can do your changes.
### Step-by-Step process:
#### 1. Import Package
This method imports various components from the superagentx library.
```python theme={null}
from superagentx.agent import Agent
from superagentx.agentxpipe import AgentXPipe
from superagentx.engine import Engine
from superagentx.llm import LLMClient
from superagentx.memory import Memory
from superagentx.prompt import PromptTemplate
from superagentx.handler import AIHandler
```
#### 2. Define Method
This method `get_content_creator_pipe` initializes a content creation pipeline by configuring an LLM
(Large Language Model) client using OpenAI's API, for more detail about `LLMClient` refer here.
```python theme={null}
async def get_content_creator_pipe() -> AgentXPipe:
# LLM Configuration
llm_config = {
'llm_type': 'openai'
}
llm_client = LLMClient(llm_config=llm_config)
```
#### 3. Memory
This line enables `memory` functionality in the pipeline by configuring a `Memory` object that created `llm_client`
for storing and retrieving, for more detail about `Memory` refer here.
```python theme={null}
# Enable Memory
memory = Memory(memory_config={"llm_client": llm_client})
```
#### 4. Handler
This block initializes AIHandler which is configured with the llm\_client to handle AI-driven operations, for more details
refer here.
```python theme={null}
# Example
ai_handler = AIHandler(
llm=llm_client
)
```
#### 5. PromptTemplate
This line creates an instance of PromptTemplate, which is used to define and manage prompt structures for the language model,
for more detail refer here.
```python theme={null}
# Prompt Template
prompt_template = PromptTemplate()
```
#### 6. Engine
This line initializes an `Engine` object for the `handlers`, combining it with the `llm_client` and `prompt_template`
to process tasks with the specified handler and prompt structure, for more details refer here.
```python theme={null}
# Example - Engine(s)
ai_engine = Engine(
handler=ai_handler,
llm=llm_client,
prompt_template=prompt_template
)
```
#### 7. Agent
This line creates an `Agent` with the role and goal to generate a list of URLs, and configuring it with the `llm_client`,
`prompt_template`, and `engine` to perform, for more details about `engine` refer here.
```python theme={null}
# Create Agent with ai Engines execute
ai_agent = Agent(
name='AI Agent',
goal="Summarize the given content",
role="You are the summarization.",
llm=llm_client,
prompt_template=prompt_template,
engines=[ai_engine]
)
```
#### 8. AgentXPipe
This line sets up an AgentXPipe, combining multiple agents (serper\_agent, crawler\_agent, and ai\_agent)
with shared memory to enable coordinated processing and communication between the agents, for more details
refer here.
```python theme={null}
# Pipe Interface to send it to public accessible interface (Cli Console / WebSocket / Restful API)
pipe = AgentXPipe(
agents=[ai_agent],
memory=memory
)
```
### All Together
```python theme={null}
from superagentx.agent import Agent
from superagentx.agentxpipe import AgentXPipe
from superagentx.engine import Engine
from superagentx.llm import LLMClient
from superagentx.memory import Memory
from superagentx.prompt import PromptTemplate
from superagentx.handler import AIHandler
async def get_content_creator_pipe() -> AgentXPipe:
# LLM Configuration
llm_config = {
'llm_type': 'openai'
}
llm_client = LLMClient(llm_config=llm_config)
# Enable Memory
memory = Memory(memory_config={"llm_client": llm_client})
ai_handler = AIHandler(
llm=llm_client
)
# Prompt Template
prompt_template = PromptTemplate()
# Example - Engine(s)
ai_engine = Engine(
handler=ai_handler,
llm=llm_client,
prompt_template=prompt_template
)
# Create Agent with ai Engines execute
ai_agent = Agent(
name='AI Agent',
goal="Summarize the given content",
role="You are the summarization.",
llm=llm_client,
prompt_template=prompt_template,
engines=[ai_engine]
)
# Pipe Interface to send it to public accessible interface (Cli Console / WebSocket / Restful API)
pipe = AgentXPipe(
agents=[ai_agent],
memory=memory
)
return pipe
```
### Run the Application
Once everything is set up, you can run your application with this command:
```shell theme={null}
# Console Pipe
$ python3 Content Creator/iopipe.py
or
# WebSocket Pipe
$ python3 Content Creator/wspipe.py
or
# RestAPI Pipe(FastAPI)
$ python3 Content Creator/restpipe.py
```
This will start your SuperAgentX application and execute the tasks you've configured.
#### Run the Application with `Verbose`
```shell theme={null}
# Console Pipe
$ VERBOSE=1 python3 Content Creator/iopipe.py
or
# Console Pipe
$ DEBUGGING=1 python3 Content Creator/iopipe.py
```
This is applicable for `wspipe.py` and `restpipe.py`. Even with your custom implementation.
Valid options are `1`, `True`, `true` and `TRUE`
### Result
```shell theme={null}
Enter your search here: Give me summary about Hon Solo
Result:
"Han Solo is a fictional character in the 'Star Wars' franchise, known for his quick
wit and roguish charm as a smuggler. He becomes a key figure in the Rebel Alliance,
aiding in significant battles like Yavin and Endor. Portrayed by Harrison Ford,
Han transforms from a self-interested rogue to a dedicated ally, playing a
crucial role in the storyline with connections to Princess Leia and Luke Skywalker."
Reason:: The summary provides an accurate and concise overview of Han Solo's
character based on the given output context.
Goal Satisfied: True
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ End โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
```
# Rest API
Source: https://docs.superagentx.ai/restapi
This documentation provides an overview of the code used to implement a RESTful API server for the pipeline using
the SuperagentX library and FastAPI. The server processes incoming requests, authenticates users via API tokens,
and handles search queries related to the trip planner.
## Overview
The code sets up a FastAPI server that processes search requests using the AgentXPipe class from SuperagentX.
It incorporates a custom authentication mechanism to validate client requests via an API token. The API exposes
an endpoint that allows users to submit a query and receive results from the trip planner pipeline.
## Setup Environment
```shell theme={null}
$ pip install superagentx 'fastapi[standard]'
```
## Implementation
Import the dependencies.
```python theme={null}
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException # https://fastapi.tiangolo.com/
from fastapi.security import APIKeyHeader
from superagentx.agentxpipe import AgentXPipe
from superagentx.result import GoalResult
from trip_planner.config import AUTH_TOKEN
from trip_planner.pipe import get_trip_planner_pipe
```
The lifespan function initializes the trip planner pipeline when the FastAPI application starts and clears
it when the app shuts down. The pipes dictionary is used to store the pipeline for later use in request handling.
```python theme={null}
@asynccontextmanager
async def lifespan(app: FastAPI):
pipes['trip_planner_pipe'] = await get_trip_planner_pipe()
yield
pipes.clear()
```
A custom dependency function, verify\_api\_token, checks the validity of the API token passed in the api-token header.
If the token is invalid, the function raises an HTTP 401 Unauthorized error.
```python theme={null}
async def verify_api_token(
api_token: str = Depends(APIKeyHeader(name='api-token', auto_error=False))
):
if api_token != AUTH_TOKEN:
raise HTTPException(status_code=401, detail='Invalid API Token!')
```
The main API application (ecom\_app) is created using FastAPI. The /search endpoint processes search queries
related to the trip planner. The verify\_api\_token function is included as a dependency to authenticate requests before processing them.
```python theme={null}
@ecom_app.get('/search', dependencies=[Depends(verify_api_token)])
async def search(query: str) -> list[GoalResult]:
trip_planner_pipe: AgentXPipe = pipes.get('trip_planner_pipe')
return await trip_planner_pipe.flow(query_instruction=query)
```
This RESTful API provides a simple yet powerful interface for interacting with the trip planner pipeline, ensuring
secure access via API token validation.
## Run the script
```shell theme={null}
$ fastapi dev trip_planner/trip_planner/restpipe.py
```
## Result
```shell theme={null}
(venv) โ superagentx_examples git:(dev) โ fastapi dev trip_planner/trip_planner/restpipe.py
INFO Using path trip_planner/trip_planner/restpipe.py
INFO Resolved absolute path /home/john/Projects/superagentX-examples/superagentx_examples/trip_planner/trip_planner/restpipe.py
INFO Searching for package john structure from directories with __init__.py files
INFO Importing from /home/john/Projects/superagentX-examples/superagentx_examples/trip_planner
โญโ Python package file structure โโฎ
โ โ
โ ๐ trip_planner โ
โ โโโ ๐ __init__.py โ
โ โโโ ๐ restpipe.py โ
โ โ
โฐโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฏ
INFO Importing module trip_planner.restpipe
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
INFO Found importable FastAPI app
โญโโโโโโโโโโโ Importable FastAPI app โโโโโโโโโโโโฎ
โ โ
โ from trip_planner.restpipe import ecom_app โ
โ โ
โฐโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฏ
INFO Using import string trip_planner.restpipe:ecom_app
โญโโโโโโโโโโ FastAPI CLI - Development mode โโโโโโโโโโโโฎ
โ โ
โ Serving at: http://127.0.0.1:8000 โ
โ โ
โ API docs: http://127.0.0.1:8000/docs โ
โ โ
โ Running in development mode, for production use: โ
โ โ
โ fastapi run โ
โ โ
โฐโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฏ
INFO: Will watch for changes in these directories: ['/home/john/Projects/superagentX-examples/superagentx_examples']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [8869] using WatchFiles
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
INFO: Started server process [8882]
INFO: Waiting for application startup.
INFO:watchfiles.main:4 changes detected
INFO: Application startup complete.
INFO:watchfiles.main:4 changes detected
```
# Storage
Source: https://docs.superagentx.ai/storage
## Data Storage & Persistence
The system supports multiple data stores to handle both lightweight development workflows and production-scale deployments.
> **SQLite** is used for local development and simple, single-node workflows where minimal setup and portability are required.
> **PostgreSQL** is recommended for production environments, supporting multi-tenant architectures, higher concurrency,
> and robust data integrity.
The data store persists all critical operational information, including:
> Workflow state to enable recovery, retries, and resumable execution
> Agent decisions for traceability and debugging
> Human approvals to support human-in-the-loop workflows
> Tool outputs generated during task execution
> Audit logs for compliance, monitoring, and observability
> Context and memory snapshots to maintain continuity across agent runs
This persistence layer ensures reliability, transparency, and full lifecycle visibility across all agent workflows.
# Voice Pipe
Source: https://docs.superagentx.ai/voice_pipe
## Voice
The VoicePipe module provides a console-based interface for utilizing the SuperAgentX voice-to-text pipelines.
It supports various backend implementations, including OpenAI Whisper and AWS Transcribe, offering flexible and
efficient voice-to-text conversion.
### Dependency
```shell theme={null}
$ sudo apt install portaudio19-dev
```
## Implementation
### OpenAI Whisper Integration
The OpenAI Whisper integration uses the WhisperPipe class from superagentx.pipeimpl.openaivoicepipe.
It processes voice-to-text requests with the following steps:
> Set up a OpenAI API key as an environmental variable and run the following code.
```shell theme={null}
export OPENAI_API_KEY = "**************************"
```
### Installation:
```shell theme={null}
$ sudo apt-get install ffmpeg
$ pip install openai-whisper NumPy=="2.1"
```
```python openaivoicepipe theme={null}
import asyncio
from rich import print as rprint
from superagentx.pipeimpl.openaivoicepipe import WhisperPipe
from create_pipe import get_superagentx_voice_to_text_pipe
async def main():
"""
Launches the superagentx-voice-to-text pipeline console client for processing requests and handling data.
"""
pipe = await get_superagentx_voice_to_text_pipe()
# Create IO Cli Console - Interface
io_pipe = WhisperPipe(
search_name='SuperAgentX - Voice To Text',
agentx_pipe=pipe,
read_prompt=f"\n[bold green]Enter your search here"
)
await io_pipe.start()
if __name__ == '__main__':
asyncio.run(main())
```
### Result of openaivoicepipe
```shell theme={null}
(venv) โ pipe git:(update_docs) โ python3.12 test_openai_iopipe_create_policy.py
Transcribing audio...
/home/vetharupini/Projects/superagentX/venv/lib/python3.12/site-packages/whisper/transcribe.py:126: UserWarning: FP16 is not supported on CPU; using FP32 instead
warnings.warn("FP16 is not supported on CPU; using FP32 instead")
Transcription successful: frames that I associated with with the
( โ ) Transcribing audio...
( โ ) Transcribing audio...
( โ ) Transcribing audio...
( โ ) Transcribing audio...
( โ ) Transcribing audio...
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
( โ ) Transcribing audio...
( โ) Transcribing audio...
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
( โ ) Transcribing audio...
( โ ) Transcribing audio...
( โ ) Transcribing audio...
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
Result:
"To create a single ticket in the HubSpot CRM, I'll need some specific details from you. Please provide the following information for the ticket:\n\n1. **Subject**: What is the subject or title of the ticket?\n2. **Content**: Please describe the content or body of the ticket.\n3. **Pipeline**: Which pipeline does this ticket belong to? (Provide the ID or default to 0 if unknown)\n4. **Pipeline Stage**: What is the stage of the pipeline for this ticket? (Provide the ID or default to 1 if unknown)\n5. **Source From**: Where is this ticket originating from? (e.g., \"EMAIL\", \"Chat\", etc., or default to \"EMAIL\")\n6. **Priority**: What is the priority level of the ticket? (e.g., \"Low\", \"Medium\", \"High\", or default to \"Low\")\n\nOnce you provide these details, I can create the ticket for you."
Reason:: The Output_Context provides a clear set of instructions to gather specific details required to create a single ticket in the HubSpot CRM.
Goal Satisfied: True
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ End โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
```
### AWS Transcribe Integration
The AWS Transcribe integration uses the AWSVoicePipe class from superagentx.pipeimpl.awsvoicepipe. It includes
region-specific configurations and follows these steps:
> Set up a AWS Access key ID as an environmental variable and run the following code.
```shell theme={null}
export AWS_ACCESS_KEY=***********************
export AWS_REGION=******
export AWS_SECRET_KEY=*****************************
```
```python awsvoicepipe theme={null}
import asyncio
from rich import print as rprint
from superagentx.pipeimpl.awsvoicepipe import AWSVoicePipe
from create_pipe import get_superagentx_voice_to_text_pipe
async def main():
"""
Launches the superagentx-voice-to-text pipeline console client for processing requests and handling data.
"""
pipe = await get_superagentx_voice_to_text_pipe()
# Create IO Cli Console - Interface
io_pipe = AWSVoicePipe(
search_name='SuperAgentX - Voice To Text',
agentx_pipe=pipe,
read_prompt=f"\n[bold green]Enter your search here",
region="us-east-1"
)
await io_pipe.start()
if __name__ == '__main__':
asyncio.run(main())
```
### Result of AWSVoicePipe
```shell theme={null}
(venv) โ pipe git:(update_docs) โ python3.12 test_aws_iopipe_create_policy.py
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
No input received within 15 seconds, stopping the stream.
Final transcriptions: ['you can get', 'you can get me the', 'you can get me the contact', 'you can get me the contact name', 'you can get me the contact name from', 'you can get me the contact name from HubSpo', 'you can get me the contact name from HubSpot.', 'me.', 'People', 'People', 'People', 'People', 'People.']
Question Asked : you can get me the contact name from HubSpot.
INFO:superagentx.agentxpipe:Pipe AgentXPipe-b0e8c00d02c74a028d382407b0e4f563 starting...
( โ ) Searching...
( โ ) Searching...
( โ ) Searching...
( โ ) Searching...
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/chat/completions "HTTP/1.1 200 OK"
( โ ) Searching...
( โ ) Searching...
( โ ) Searching...
INFO:httpx:HTTP Request: POST https://api.openai.com/v1/embeddings "HTTP/1.1 200 OK"
Result:
"Contact names available from the given data: Arul Vivek, Arul I, Ram Kumar"
Reason:: The output context provides the contact names from HubSpot needed to create a new ticket.
Goal Satisfied: True
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ End โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
```
# WebSocket
Source: https://docs.superagentx.ai/websocket
This documentation provides an overview of the implementation of a WebSocket server for the trip planner pipeline using
SuperagentX. The server processes incoming WebSocket requests, authenticates users via query parameters, and handles
real-time data processing through the pipeline.
## Overview
The code sets up a WebSocket server using the WSPipe class from SuperagentX. It integrates a custom authentication
mechanism to validate client requests based on a token passed in the query parameters.
## Setup Environment
```shell theme={null}
$ pip install superagentx
```
## Implementation
Import the dependencies.
```python theme={null}
import asyncio
import http
import urllib.parse
from rich import print as rprint
from superagentx.pipeimpl.wspipe import WSPipe # https://websockets.readthedocs.io/en/stable/
from trip_planner.config import AUTH_TOKEN
from trip_planner.pipe import get_trip_planner_pipe
```
The query\_param\_auth function authenticates users based on a token included in the WebSocket request's query parameters.
It verifies the token's presence and validity against the predefined AUTH\_TOKEN. If the token is missing or invalid,
the server responds with an HTTP 401 Unauthorized status.
```python theme={null}
async def query_param_auth(connection, request):
"""Authenticate user from token in query parameter."""
query = urllib.parse.urlparse(request.path).query
params = urllib.parse.parse_qs(query)
values = params.get('token', [])
if values:
token = values[0]
if token is None:
return connection.respond(http.HTTPStatus.UNAUTHORIZED, "Missing token\n")
if token != AUTH_TOKEN:
return connection.respond(http.HTTPStatus.UNAUTHORIZED, "Invalid token\n")
```
The main function initializes the trip planner pipeline using get\_trip\_planner\_pipe. It then creates a WSPipe object,
which serves as the WebSocket server for processing requests. The server uses query\_param\_auth for authentication and
dynamically handles incoming data.
```python theme={null}
async def main():
"""
Launches the trip_planner pipeline websocket server for processing requests and handling data.
"""
pipe = await get_trip_planner_pipe()
ws_pipe = WSPipe(
search_name='SuperAgentX trip_planner Websocket Server',
agentx_pipe=pipe,
process_request=query_param_auth
)
await ws_pipe.start()
if __name__ == '__main__':
try:
asyncio.run(main())
except (KeyboardInterrupt, asyncio.CancelledError):
rprint("\nUser canceled the [bold yellow][i]pipe[/i]!")
```
This WebSocket server provides a secure and efficient interface for real-time interaction with the trip planner pipeline.
By leveraging token-based authentication, it ensures that only authorized users can access the service.
## Run the script
```shell theme={null}
$ python3 wspipe.py
```
## Result
```shell theme={null}
(venv) โ trip_planner git:(dev) โ python3 wspipe.py
Warning: Synchronous WebCrawler is not available. Install crawl4ai[sync] for synchronous support. However, please note that the synchronous version will be deprecated soon.
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ SuperAgentX trip_planner Websocket Server โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
๐ Starting SuperagentX websocket server
๐ Host: localhost
๐ Port: 8765
INFO:websockets.server:server listening on 127.0.0.1:8765
```