select_ai.agent also provides async interfaces to be used with
async / await keywords
1. AsyncTool¶
- class select_ai.agent.AsyncTool(tool_name: str | None = None, description: str | None = None, attributes: ToolAttributes | None = None)¶
- async classmethod create_built_in_tool(tool_name: str, tool_params: ToolParams, tool_type: ToolType, description: str | None = None, replace: bool | None = False) AsyncTool¶
Register a built-in tool
- Parameters:
tool_name (str) – The name of the tool
tool_params (select_ai.agent.ToolParams) – Parameters required by built-in tool
tool_type (select_ai.agent.ToolType) – The built-in tool type
description (str) – Description of the tool
replace (bool) – Whether to replace the existing tool. Default value is False
- Returns:
select_ai.agent.Tool
- async classmethod create_email_notification_tool(tool_name: str, credential_name: str, recipient: str, sender: str, smtp_host: str, description: str | None, replace: bool = False) AsyncTool¶
Register an email notification tool
- Parameters:
tool_name (str) – The name of the tool
credential_name (str) – The name of the credential
recipient (str) – The recipient of the email
sender (str) – The sender of the email
smtp_host (str) – The SMTP host of the email server
description (str) – The description of the tool
replace (bool) – Whether to replace the existing tool. Default value is False
- Returns:
select_ai.agent.Tool
- async classmethod create_pl_sql_tool(tool_name: str, function: str, description: str | None = None, replace: bool = False) AsyncTool¶
Create a custom tool to invoke PL/SQL procedure or function
- Parameters:
tool_name (str) – The name of the tool
function (str) – The name of the PL/SQL procedure or function
description (str) – The description of the tool
replace (bool) – Whether to replace existing tool. Default value is False
- async classmethod create_rag_tool(tool_name: str, profile_name: str, description: str | None = None, replace: bool = False) AsyncTool¶
Register a RAG tool, which will use a VectorIndex linked AI Profile
- Parameters:
tool_name (str) – The name of the tool
profile_name (str) – The name of the profile to use for Vector Index based RAG
description (str) – The description of the tool
replace (bool) – Whether to replace existing tool. Default value is False
- async classmethod create_slack_notification_tool(tool_name: str, credential_name: str, slack_channel: str, description: str | None = None, replace: bool = False) AsyncTool¶
Register a Slack notification tool
- Parameters:
tool_name (str) – The name of the Slack notification tool
credential_name (str) – The name of the Slack credential
slack_channel (str) – The name of the Slack channel
description (str) – The description of the Slack notification tool
replace (bool) – Whether to replace existing tool. Default value is False
- async classmethod create_sql_tool(tool_name: str, profile_name: str, description: str | None = None, replace: bool = False) AsyncTool¶
Register a SQL tool to perform natural language to SQL translation
- Parameters:
tool_name (str) – The name of the tool
profile_name (str) – The name of the profile to use for SQL translation
description (str) – The description of the tool
replace (bool) – Whether to replace existing tool. Default value is False
- async classmethod create_websearch_tool(tool_name: str, credential_name: str, description: str | None, replace: bool = False) AsyncTool¶
Register a built-in websearch tool to search information on the web
- Parameters:
tool_name (str) – The name of the tool
credential_name (str) – The name of the credential object storing OpenAI credentials
description (str) – The description of the tool
replace (bool) – Whether to replace the existing tool
- async delete(force: bool = False)¶
Delete AI Tool from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async disable()¶
Disable AI Tool
- async enable()¶
Enable AI Tool
- async classmethod fetch(tool_name: str) AsyncTool¶
Fetch AI Tool attributes from the Database and build a proxy object in the Python layer
- Parameters:
tool_name (str) – The name of the AI Task
- Returns:
select_ai.agent.Tool
- Raises:
select_ai.errors.AgentToolNotFoundError – If the AI Tool is not found
- classmethod list(tool_name_pattern: str = '.*') AsyncGenerator[AsyncTool, None]¶
List AI Tools
- Parameters:
tool_name_pattern (str) – Regular expressions can be used to specify a pattern. Function REGEXP_LIKE is used to perform the match. Default value is “.*” i.e. match all tool name.
- Returns:
Iterator[Tool]
- async set_attribute(attribute_name: str, attribute_value: Any) None¶
Set the attribute of the AI Agent tool specified by attribute_name and attribute_value.
- async set_attributes(attributes: ToolAttributes) None¶
Set the attributes of the AI Agent tool
1.1. Create Tool¶
The following example shows async creation of an AI agent tool to perform natural language translation to SQL using an OCI AI profile
import asyncio
import os
from pprint import pformat
import select_ai
import select_ai.agent
from select_ai.agent import AsyncTool, ToolAttributes
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
profile_attributes = select_ai.ProfileAttributes(
credential_name="my_oci_ai_profile_key",
object_list=[
{"owner": user, "name": "MOVIE"},
{"owner": user, "name": "ACTOR"},
{"owner": user, "name": "DIRECTOR"},
],
provider=select_ai.OCIGenAIProvider(
region="us-chicago-1",
oci_apiformat="GENERIC",
model="meta.llama-4-maverick-17b-128e-instruct-fp8",
),
)
profile = await select_ai.AsyncProfile(
profile_name="LLAMA_4_MAVERICK",
attributes=profile_attributes,
description="MY OCI AI Profile",
replace=True,
)
# Create a tool which uses the OCI AI Profile to
# perform natural language SQL translation
sql_tool = await AsyncTool.create_sql_tool(
tool_name="MOVIE_SQL_TOOL",
description="My Select AI MOVIE SQL agent tool",
profile_name="LLAMA_4_MAVERICK",
replace=True,
)
print(sql_tool.tool_name)
print(pformat(sql_tool.attributes))
asyncio.run(main())
output:
MOVIE_SQL_TOOL
ToolAttributes(instruction=None,
function=None,
tool_params=SQLToolParams(_REQUIRED_FIELDS=None,
credential_name=None,
endpoint=None,
notification_type=None,
profile_name='oci_ai_profile',
recipient=None,
sender=None,
slack_channel=None,
smtp_host=None),
tool_inputs=None,
tool_type=<ToolType.SQL: 'SQL'>)
1.2. List Tools¶
import os
import select_ai
from select_ai.agent import AsyncTool
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
async for tool in AsyncTool.list():
print(tool.tool_name)
asyncio.run(main())
output:
WEB_SEARCH_TOOL
MOVIE_SQL_TOOL
LLM_CHAT_TOOL
2. AsyncTask¶
- class select_ai.agent.AsyncTask(task_name: str | None = None, description: str | None = None, attributes: TaskAttributes | None = None)¶
select_ai.agent.AsyncTask class lets you create, delete, enable, disable and list AI Tasks asynchronously
- Parameters:
task_name (str) – The name of the AI task
description (str) – Optional description of the AI task
attributes (select_ai.agent.TaskAttributes) – AI task attributes
- async create(enabled: bool | None = True, replace: bool | None = False)¶
Create a task that a Select AI agent can include in its reasoning process
- Parameters:
enabled (bool) – Whether the AI Task should be enabled. Default value is True.
replace (bool) – Whether the AI Task should be replaced. Default value is False.
- async delete(force: bool = False)¶
Delete AI Task from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async disable()¶
Disable AI Task
- async enable()¶
Enable AI Task
- async classmethod fetch(task_name: str) AsyncTask¶
Fetch AI Task attributes from the Database and build a proxy object in the Python layer
- Parameters:
task_name (str) – The name of the AI Task
- Returns:
select_ai.agent.Task
- Raises:
select_ai.errors.AgentTaskNotFoundError – If the AI Task is not found
- classmethod list(task_name_pattern: str | None = '.*') AsyncGenerator[AsyncTask, None]¶
List AI Tasks
- Parameters:
task_name_pattern (str) – Regular expressions can be used to specify a pattern. Function REGEXP_LIKE is used to perform the match. Default value is “.*” i.e. match all tasks.
- Returns:
AsyncGenerator[Task]
- async set_attribute(attribute_name: str, attribute_value: Any)¶
Set a single AI Task attribute specified using name and value
- Parameters:
attribute_name (str) – The name of the AI Task attribute
attribute_value (str) – The value of the AI Task attribute
- async set_attributes(attributes: TaskAttributes)¶
Set AI Task attributes
- Parameters:
attributes (select_ai.agent.TaskAttributes) – Multiple attributes can be specified by passing a TaskAttributes object
2.1. Create Task¶
In the following task, we use the MOVIE_SQL_TOOL created in the
previous step
import asyncio
import os
from pprint import pformat
import select_ai
import select_ai.agent
from select_ai.agent import AsyncTask, TaskAttributes
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
task = AsyncTask(
task_name="ANALYZE_MOVIE_TASK",
description="Search for movies in the database",
attributes=TaskAttributes(
instruction="Help the user with their request about movies. "
"User question: {query}. "
"You can use SQL tool to search the data from database",
tools=["MOVIE_SQL_TOOL"],
enable_human_tool=False,
),
)
await task.create(replace=True)
print(task.task_name)
print(pformat(task.attributes))
asyncio.run(main())
output:
ANALYZE_MOVIE_TASK
TaskAttributes(instruction='Help the user with their request about movies. '
'User question: {query}. You can use SQL tool to '
'search the data from database',
tools=['MOVIE_SQL_TOOL'],
input=None,
enable_human_tool=False)
2.2. List Tasks¶
import asyncio
import os
import select_ai
from select_ai.agent import AsyncTask
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
async for task in AsyncTask.list():
print(task.task_name)
asyncio.run(main())
output:
WEB_SEARCH_TASK
ANALYZE_MOVIE_TASK
3. AsyncAgent¶
- class select_ai.agent.AsyncAgent(agent_name: str | None = None, description: str | None = None, attributes: AgentAttributes | None = None)¶
select_ai.agent.AsyncAgent class lets you create, delete, enable, disable and list AI agents asynchronously
- Parameters:
agent_name (str) – The name of the AI Agent
description (str) – Optional description of the AI agent
attributes (select_ai.agent.AgentAttributes) – AI agent attributes
- async create(enabled: bool | None = True, replace: bool | None = False)¶
Register a new AI Agent within the Select AI framework
- Parameters:
enabled (bool) – Whether the AI Agent should be enabled. Default value is True.
replace (bool) – Whether the AI Agent should be replaced. Default value is False.
- async delete(force: bool | None = False)¶
Delete AI Agent from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async disable()¶
Disable AI Agent
- async enable()¶
Enable AI Agent
- async classmethod fetch(agent_name: str) AsyncAgent¶
Fetch AI Agent attributes from the Database and build a proxy object in the Python layer
- Parameters:
agent_name (str) – The name of the AI Agent
- Returns:
select_ai.agent.Agent
- Raises:
select_ai.errors.AgentNotFoundError – If the AI Agent is not found
- classmethod list(agent_name_pattern: str | None = '.*') AsyncGenerator[AsyncAgent, None]¶
List AI agents matching a pattern
- Parameters:
agent_name_pattern (str) – Regular expressions can be used to specify a pattern. Function REGEXP_LIKE is used to perform the match. Default value is “.*” i.e. match all agent names.
- Returns:
AsyncGenerator[AsyncAgent]
- async set_attribute(attribute_name: str, attribute_value: Any) None¶
Set a single AI Agent attribute specified using name and value
- async set_attributes(attributes: AgentAttributes) None¶
Set AI Agent attributes
- Parameters:
attributes (select_ai.agent.AgentAttributes) – Multiple attributes can be specified by passing an AgentAttributes object
3.1. Create Agent¶
import asyncio
import os
import select_ai
from select_ai.agent import (
AgentAttributes,
AsyncAgent,
)
async def main():
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
await select_ai.async_connect(user=user, password=password, dsn=dsn)
agent_attributes = AgentAttributes(
profile_name="LLAMA_4_MAVERICK",
role="You are an AI Movie Analyst. "
"Your can help answer a variety of questions related to movies. ",
enable_human_tool=False,
)
agent = AsyncAgent(
agent_name="MOVIE_ANALYST",
attributes=agent_attributes,
)
await agent.create(enabled=True, replace=True)
print("Created Agent:", agent)
asyncio.run(main())
output:
Created Agent: Agent(agent_name=MOVIE_ANALYST,
attributes=AgentAttributes(profile_name='LLAMA_4_MAVERICK',
role='You are an AI Movie Analyst.
Your can help answer a variety of questions related to movies. ',
enable_human_tool=False), description=None)
3.2. List Agents¶
import asyncio
import os
import select_ai
from select_ai.agent import AsyncAgent
async def main():
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
await select_ai.async_connect(user=user, password=password, dsn=dsn)
async for agent in AsyncAgent.list():
print(agent.agent_name)
asyncio.run(main())
output:
WEB_SEARCH_AGENT
MOVIE_ANALYST
4. AsyncTeam¶
- class select_ai.agent.AsyncTeam(team_name: str, attributes: TeamAttributes, description: str | None = None)¶
A Team of AI agents work together to accomplish tasks select_ai.agent.Team class lets you create, delete, enable, disable and list AI Tasks.
- Parameters:
team_name (str) – The name of the AI team
description (str) – Optional description of the AI team
attributes (select_ai.agent.TeamAttributes) – AI team attributes
- async create(enabled: bool | None = True, replace: bool | None = False)¶
Create a team of AI agents that work together to accomplish tasks.
- Parameters:
enabled (bool) – Whether the AI agent team should be enabled. Default value is True.
replace (bool) – Whether the AI agent team should be replaced. Default value is False.
- async delete(force: bool | None = False)¶
Delete an AI agent team from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async disable()¶
Disable the AI agent team
- async enable()¶
Enable the AI agent team
- async classmethod fetch(team_name: str) AsyncTeam¶
Fetch AI Team attributes from the Database and build a proxy object in the Python layer
- Parameters:
team_name (str) – The name of the AI Team
- Returns:
select_ai.agent.Team
- Raises:
select_ai.errors.AgentTeamNotFoundError – If the AI Team is not found
- classmethod list(team_name_pattern: str | None = '.*') AsyncGenerator[AsyncTeam, None]¶
List AI Agent Teams
- Parameters:
team_name_pattern (str) – Regular expressions can be used to specify a pattern. Function REGEXP_LIKE is used to perform the match. Default value is “.*” i.e. match all teams.
- Returns:
Iterator[Team]
- async run(prompt: str = None, params: Mapping = None)¶
Start a new AI agent team or resume a paused one that is waiting for human input. If you provide an existing process ID and the associated team process is in the WAITING_FOR_HUMAN state, the function resumes the workflow using the input you provide as the human response
- Parameters:
prompt (str) – Optional prompt for the user. If the task is in the RUNNING state, the input acts as a placeholder for the {query} in the task instruction. If the task is in the WAITING_FOR_HUMAN state, the input serves as the human response.
params (Mapping[str, str]) –
Optional parameters for the task. Currently, the following parameters are supported:
conversation_id: Identifies the conversation session associated
with the agent team
variables: key-value pairs that provide additional input to the agent team.
- async set_attribute(attribute_name: str, attribute_value: Any) None¶
Set the attribute of the AI Agent team specified by attribute_name and attribute_value.
- async set_attributes(attributes: TeamAttributes) None¶
Set the attributes of the AI Agent team
4.1. Run Team¶
import asyncio
import os
import uuid
import select_ai
from select_ai.agent import (
AsyncTeam,
TeamAttributes,
)
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
team = AsyncTeam(
team_name="MOVIE_AGENT_TEAM",
attributes=TeamAttributes(
agents=[{"name": "MOVIE_ANALYST", "task": "ANALYZE_MOVIE_TASK"}],
process="sequential",
),
)
await team.create(enabled=True, replace=True)
print(
await team.run(
prompt="Could you list the movies in the database?",
params={"conversation_id": str(uuid.uuid4())},
)
)
asyncio.run(main())
output:
The database contains 100 movies with various titles, genres, and release
dates. The list includes a wide range of genres such as Action, Comedy, Drama,
Thriller, Romance, Adventure, Mystery, Sci-Fi, Historical, Biography, War,
Sports, Music, Documentary, Animated, Fantasy, Horror, Western, Family,
and more. The release dates are primarily in January and February of 2019.
Here is a summary of the movies:
1. Action Movie (Action, 2019-01-01)
2. Comedy Film (Comedy, 2019-01-02)
3. Drama Series (Drama, 2019-01-03)
4. Thriller Night (Thriller, 2019-01-04)
5. Romance Story (Romance, 2019-01-05)
6. Adventure Time (Adventure, 2019-01-06)
7. Mystery Solver (Mystery, 2019-01-07)
8. Sci-Fi World (Sci-Fi, 2019-01-08)
9. Historical Epic (Historical, 2019-01-09)
10. Biographical (Biography, 2019-01-10)
... (list continues up to 100 movies)
4.2. List Teams¶
import asyncio
import os
import select_ai
from select_ai.agent import AsyncTeam
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
async for team in AsyncTeam.list():
print(team.team_name)
asyncio.run(main())
output:
WEB_SEARCH_TEAM
MOVIE_AGENT_TEAM
5. Async AI agent examples¶
5.1. Web Search Agent using OpenAI’s GPT model¶
import asyncio
import os
import select_ai
from select_ai.agent import (
AgentAttributes,
AsyncAgent,
AsyncTask,
AsyncTeam,
AsyncTool,
TaskAttributes,
TeamAttributes,
)
OPEN_AI_CREDENTIAL_NAME = "OPENAI_CRED"
OPEN_AI_PROFILE_NAME = "OPENAI_PROFILE"
SELECT_AI_AGENT_NAME = "WEB_SEARCH_AGENT"
SELECT_AI_TASK_NAME = "WEB_SEARCH_TASK"
SELECT_AI_TOOL_NAME = "WEB_SEARCH_TOOL"
SELECT_AI_TEAM_NAME = "WEB_SEARCH_TEAM"
USER_QUERIES = {
"d917b055-e8a1-463a-a489-d4328a7b2210": "What are the key features for the product highlighted at "
"this URL https://www.oracle.com/artificial-intelligence/database-machine-learning",
"c2e3ff20-f56d-40e7-987c-cc72740c75a5": "What is the main topic at this URL https://www.oracle.com/artificial-intelligence/database-machine-learning",
"25e23a25-07b9-4ed7-be11-f7e5e445d286": "What is the main topic at this URL https://openai.com",
}
# connect
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
# Create Open AI credential
await select_ai.async_create_credential(
credential={
"credential_name": OPEN_AI_CREDENTIAL_NAME,
"username": "OPENAI",
"password": os.getenv("OPEN_AI_API_KEY"),
},
replace=True,
)
print("Created credential: ", OPEN_AI_CREDENTIAL_NAME)
# # Create Open AI Profile
profile = await select_ai.AsyncProfile(
profile_name=OPEN_AI_PROFILE_NAME,
attributes=select_ai.ProfileAttributes(
credential_name=OPEN_AI_CREDENTIAL_NAME,
provider=select_ai.OpenAIProvider(model="gpt-4.1"),
),
description="My Open AI Profile",
replace=True,
)
print("Created profile: ", OPEN_AI_PROFILE_NAME)
# Create an AI Agent team
team = AsyncTeam(
team_name=SELECT_AI_TEAM_NAME,
attributes=TeamAttributes(
agents=[
{"name": SELECT_AI_AGENT_NAME, "task": SELECT_AI_TASK_NAME}
]
),
)
await team.create(replace=True)
# Agent
agent = AsyncAgent(
agent_name=SELECT_AI_AGENT_NAME,
attributes=AgentAttributes(
profile_name=OPEN_AI_PROFILE_NAME,
enable_human_tool=False,
role="You are a specialized web search agent that can access web page "
"contents and respond to questions based on its content.",
),
)
await agent.create(replace=True)
# Task
task = AsyncTask(
task_name=SELECT_AI_TASK_NAME,
attributes=TaskAttributes(
instruction="Answer the user question about the provided URL:{query}",
enable_human_tool=False,
tools=[SELECT_AI_TOOL_NAME],
),
)
await task.create(replace=True)
# Tool
web_search_tool = await AsyncTool.create_websearch_tool(
tool_name=SELECT_AI_TOOL_NAME,
credential_name=OPEN_AI_CREDENTIAL_NAME,
description="Web Search Tool using OpenAI",
replace=True,
)
print("Created tool: ", SELECT_AI_TOOL_NAME)
# Run the Agent Team
for conversation_id, prompt in USER_QUERIES.items():
response = await team.run(
prompt=prompt, params={"conversation_id": conversation_id}
)
print(response)
asyncio.run(main())
output:
Created credential: OPENAI_CRED
Created profile: OPENAI_PROFILE
Created tool: WEB_SEARCH_TOOL
The key features of Oracle Database Machine Learning, as highlighted on the
Oracle website, include:
- In-database machine learning: Build, train, and deploy machine learning
models directly inside the Oracle Database, eliminating the need to move
data.
- Support for multiple languages: Use SQL, Python, and R for machine
learning tasks, allowing flexibility for data scientists and developers.
- Automated machine learning (AutoML): Automates feature selection, model
selection, and hyperparameter tuning to speed up model development.
- Scalability and performance: Utilizes Oracle Database’s scalability,
security, and high performance for machine learning workloads.
- Integration with Oracle Cloud: Seamlessly integrates with Oracle
Cloud Infrastructure for scalable and secure deployment.
- Security and governance: Inherits Oracle Database’s robust security,
data privacy, and governance features.
- Prebuilt algorithms: Offers a wide range of in-database algorithms for
classification, regression, clustering, anomaly detection, and more.
- No data movement: Keeps data secure and compliant by performing
analytics and machine learning where the data resides.
These features enable organizations to operationalize machine learning at
scale, improve productivity, and maintain data security and compliance.
The main topic at the URL https://www.oracle.com/artificial-intelligence/database-machine-learning
is Oracle's database machine learning capabilities, specifically how Oracle
integrates artificial intelligence and machine learning features directly
into its database products. The page highlights how users can leverage these
built-in AI and ML tools to analyze data, build predictive models, and enhance
business applications without moving data outside the Oracle Database
environment.
The main topic of the website https://openai.com is artificial
intelligence research and development. OpenAI focuses on creating and
promoting advanced AI technologies, including products like ChatGPT, and
provides information about their research, products, and mission to ensure
that artificial general intelligence benefits all of humanity.