select_ai.agent also provides async interfaces to be used with
async / await keywords. Use these classes in applications that already
use asyncio and select_ai.async_connect() or
select_ai.create_pool_async().
The history API follows the same pattern. AsyncTeamHistory,
AsyncTaskHistory, and AsyncToolHistory query only the current user’s
history views and yield typed events newest first.
See Async agent history for the complete API reference and sample.
from select_ai.agent import AsyncToolHistory
async for call in AsyncToolHistory.list(limit=10):
print(call.tool_name, call.output)
The async sample retrieves a team’s latest execution and uses its
team_exec_id to retrieve the associated task and tool history.
The async agent object model mirrors the synchronous agent object model:
Sync class |
Async class |
|---|---|
|
|
|
|
|
|
|
|
Create or reuse the same database objects as the synchronous APIs. Async methods must be awaited, and async list methods return async iterators.
await select_ai.async_connect(user=user, password=password, dsn=dsn)
async for tool in select_ai.agent.AsyncTool.list():
print(tool.tool_name)
Tools, tasks, agents, and teams are database objects. Use replace=True when
you want to recreate an existing object with the same name, and force=True
when cleanup should succeed even if the object does not exist.
Tool Type |
AsyncTool Class Method |
Arguments |
|---|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Notification and web search tools require credentials and network access for the external service. SQL and RAG tools require existing Select AI profiles.
Tool selection follows the same guidance as Agent: use SQL tools for database questions, RAG tools for vector-index-backed content, notification tools for Slack or email, web search tools for public web content, and PL/SQL tools for application-specific database logic.
1. AsyncTool¶
- class select_ai.agent.AsyncTool(tool_name: str | None = None, description: str | None = None, attributes: ToolAttributes | None = None)¶
- async create(enabled: bool | None = True, replace: bool | None = False)¶
Create an AI Tool in the database
- async classmethod create_built_in_tool(tool_name: str, tool_params: ToolParams, tool_type: ToolType, description: str | None = None, replace: bool | None = False, instruction: str | None = None) 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.tool.ToolType) – The built-in tool type
description (str) – Description of the tool
replace (bool) – Whether to replace the existing tool. Default value is False
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- 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, subject: str | None = None, replace: bool = False, instruction: str | None = None) 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
subject (str) – Subject of the email.
replace (bool) – Whether to replace the existing tool. Default value is False
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- Returns:
select_ai.agent.Tool
- async classmethod create_pl_sql_tool(tool_name: str, function: str, description: str | None = None, replace: bool = False, instruction: str | None = None) 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
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- async classmethod create_rag_tool(tool_name: str, profile_name: str, description: str | None = None, replace: bool = False, instruction: str | None = None) 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
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- async classmethod create_slack_notification_tool(tool_name: str, credential_name: str, channel: str, description: str | None = None, replace: bool = False, instruction: str | None = None) 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
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
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- async classmethod create_sql_tool(tool_name: str, profile_name: str, description: str | None = None, replace: bool = False, instruction: str | None = None) 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
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- async classmethod create_websearch_tool(tool_name: str, credential_name: str, description: str | None, replace: bool = False, instruction: str | None = None) 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
instruction (str) – A clear, concise statement that describes what the tool should accomplish and how to do it. This text is included in the prompt sent to the LLM.
- async delete(force: bool = False)¶
Delete AI Tool from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async classmethod delete_tool(tool_name: str, force: bool = False)¶
Class method ot delete AI Tool from the database
- 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,
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
1.3. Inspect and run a tool¶
Use AsyncTool.describe_tool() to return JSON metadata for a tool, including
its function arguments. Use AsyncTool.run_tool(input) to invoke a tool
directly with an input payload.
tool = await AsyncTool.fetch("MOVIE_SQL_TOOL")
print(await tool.describe_tool())
print(await tool.run_tool('{"query": "How many movies are there?"}'))
The complete example creates a temporary PL/SQL tool, describes it, invokes it, and removes the temporary database objects:
import asyncio
import os
import uuid
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)
suffix = uuid.uuid4().hex.upper()
function_name = f"SAMPLE_CALCULATE_AGE_{suffix}"
tool_name = f"SAMPLE_AGE_TOOL_{suffix}"
async with select_ai.async_cursor() as cursor:
await cursor.execute(
f"""
CREATE OR REPLACE FUNCTION {function_name}(p_birth_date DATE)
RETURN NUMBER IS
BEGIN
RETURN TRUNC(MONTHS_BETWEEN(SYSDATE, p_birth_date) / 12);
END;
"""
)
tool = await AsyncTool.create_pl_sql_tool(
tool_name=tool_name,
function=function_name,
description="Calculate age from a birth date",
)
try:
print("Tool description:")
print(await tool.describe_tool())
print("Tool result:")
print(await tool.run_tool('{"p_birth_date":"2000-01-01"}'))
finally:
await tool.delete(force=True)
async with select_ai.async_cursor() as cursor:
await cursor.execute(f"DROP FUNCTION {function_name}")
asyncio.run(main())
The generated tool name and calculated age vary between runs. Representative output is:
output:
Tool description:
{"tool_name": "SAMPLE_AGE_TOOL_<generated suffix>", ...}
Tool result:
<calculated age in years>
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
- async delete(force: bool = False)¶
Delete AI Task from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async classmethod delete_task(task_name: str, force: bool = False)¶
Class method to delete AI Task from the database
- 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
- 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
The instruction is the main task prompt. Use placeholders such as
{query} when the user prompt should be inserted into the task. The
tools list limits which tools the agent can use for the task.
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
- async delete(force: bool | None = False)¶
Delete AI Agent from the database
- Parameters:
force (bool) – Force the deletion. Default value is False.
- async classmethod delete_agent(agent_name: str, force: bool | None = False)¶
Class method to delete AI Agent from the database
- 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 | None = None, 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.
- 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 classmethod delete_team(team_name: str, force: bool | None = False)¶
Class method to delete an AI agent team from the database
- async disable()¶
Disable the AI agent team
- async enable()¶
Enable the AI agent team
- async export(object_storage_credential_name: str | None = None, location: str | None = None, params: str | Mapping | None = None) str | None¶
Export this AI agent team specification.
If object storage details are provided, the specification is written to the given location and None is returned. Otherwise, the specification is returned as a string.
- Parameters:
object_storage_credential_name (str) – Optional credential name used to write the exported specification to object storage. Must be specified together with
location.location (str) – Optional object storage URI where the exported specification should be written. Must be specified together with
object_storage_credential_name.params (str or Mapping) – Optional export parameters. May be a JSON string or a Python mapping.
- Returns:
Exported team specification as a JSON string when exporting inline, or None when exporting to object storage.
- Return type:
str or None
- async classmethod export_team(team_name: str, object_storage_credential_name: str | None = None, location: str | None = None, params: str | Mapping | None = None) str | None¶
Export an AI agent team specification.
If object storage details are provided, the specification is written to the given location and None is returned. Otherwise, the specification is returned as a string.
- Parameters:
team_name (str) – Name of the AI agent team to export.
object_storage_credential_name (str) – Optional credential name used to write the exported specification to object storage. Must be specified together with
location.location (str) – Optional object storage URI where the exported specification should be written. Must be specified together with
object_storage_credential_name.params (str or Mapping) – Optional export parameters. May be a JSON string or a Python mapping.
- Returns:
Exported team specification as a JSON string when exporting inline, or None when exporting to object storage.
- Return type:
str or None
- 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
- async grant_access(user_or_role_name: str) None¶
Asynchronously grant a user or role access to this agent team.
- async classmethod import_team(profile_name: str, team_name: str | None = None, specification: str | Mapping | None = None, object_storage_credential_name: str | None = None, location: str | None = None, force: bool | None = False, params: str | Mapping | None = None) None¶
Import an AI agent team specification and create the associated team, agents, tasks, and tools in the database.
- Parameters:
profile_name (str) – Name of the Select AI profile to use for the imported team and agents in the target database.
team_name (str) – Optional name for the imported team. If omitted, the team name from the specification is used.
specification (str or Mapping) – Team specification to import. May be a JSON string or a Python mapping. Omit this when importing from object storage.
object_storage_credential_name (str) – Optional credential name used to read the specification from object storage. Must be specified together with
location.location (str) – Optional object storage URI of the specification to import. Must be specified together with
object_storage_credential_name.force (bool) – Whether to replace conflicting database objects during import. Default value is False.
params (str or Mapping) – Optional import parameters. May be a JSON string or a Python mapping.
- 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 revoke_access(user_or_role_name: str) None¶
Asynchronously revoke a user or role’s agent team access.
- 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. Supported keys include
conversation_id, which identifies the conversation session associated with the agent team, andvariables, which provides additional key-value 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.2. Supervised teams¶
Configure an asynchronous supervised team in the same way as a synchronous
team. Set supervisor=True on the coordinating agent and pass its name as
TeamAttributes.supervisor_agent. Keep the worker agent and task
assignments in the team’s agents list.
The database creates the supervisor task when the team is created, so
supervisor_task should not be supplied by the application. Fetch the team
after creation to read the generated value from
fetched.attributes.supervisor_task.
from select_ai.agent import (
AgentAttributes, AsyncAgent, AsyncTeam, TeamAttributes,
)
# Inside an async function, with an open async connection.
# MOVIE_ANALYST and ANALYZE_MOVIE_TASK must already exist.
supervisor = AsyncAgent(
agent_name="MOVIE_SUPERVISOR",
attributes=AgentAttributes(
profile_name="oci_ai_profile",
role="You supervise and coordinate the team.",
supervisor=True,
),
)
await supervisor.create()
team = AsyncTeam(
team_name="MOVIE_AGENT_TEAM",
attributes=TeamAttributes(
agents=[
{"name": "MOVIE_ANALYST", "task": "ANALYZE_MOVIE_TASK"}
],
process="sequential",
supervisor_agent=supervisor.agent_name,
),
)
await team.create()
Constructors build Python objects. Await create() to persist each object
in Oracle Database. The example expects new supervisor and team names; pass
replace=True to create() only when replacement is intended.
The complete asynchronous example creates the team, runs a prompt through the
supervisor workflow, and inspects its generated supervisor task and metadata.
AsyncTeam.run() starts the team workflow; the configured supervisor agent
coordinates the worker agent as part of that call.
Set the common database connection variables and SELECT_AI_PROFILE_NAME
(default oci_ai_profile) to an existing profile. Run
python samples/agent/async/team_supervisor_inspect.py from the repository
root. The script removes its generated objects and conversation during cleanup.
Configure the imports and connection values:
import asyncio
import os
import uuid
import select_ai
from select_ai.agent import (
AgentAttributes,
AsyncAgent,
AsyncTask,
AsyncTeam,
TaskAttributes,
TeamAttributes,
)
from select_ai.conversation import (
AsyncConversation,
ConversationAttributes,
)
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
profile_name = os.getenv("SELECT_AI_PROFILE_NAME", "oci_ai_profile")
Append the async function and its task, agent, and team definitions:
async def main():
suffix = uuid.uuid4().hex.upper()
await select_ai.async_connect(user=user, password=password, dsn=dsn)
task = AsyncTask(
task_name=f"SAMPLE_TASK_{suffix}",
attributes=TaskAttributes(
instruction="Answer the user's question: {query}",
tools=[],
enable_human_tool=False,
),
)
worker = AsyncAgent(
agent_name=f"SAMPLE_WORKER_{suffix}",
attributes=AgentAttributes(
profile_name=profile_name,
role="You answer user questions.",
enable_human_tool=False,
),
)
supervisor = AsyncAgent(
agent_name=f"SAMPLE_SUPERVISOR_{suffix}",
attributes=AgentAttributes(
profile_name=profile_name,
role="You supervise and coordinate the team.",
enable_human_tool=False,
supervisor=True,
),
)
team = AsyncTeam(
team_name=f"SAMPLE_TEAM_{suffix}",
attributes=TeamAttributes(
agents=[{"name": worker.agent_name, "task": task.task_name}],
process="sequential",
supervisor_agent=supervisor.agent_name,
),
)
conversation = AsyncConversation(
attributes=ConversationAttributes(
title="Supervised team sample",
description="Conversation for the supervised team sample",
)
)
Continue inside main() with creation, execution, inspection, and cleanup.
Keep this block indented as part of the function, then add the entry point:
created_objects = []
conversation_created = False
try:
await task.create(replace=True)
created_objects.append(task)
await worker.create(replace=True)
created_objects.append(worker)
await supervisor.create(replace=True)
created_objects.append(supervisor)
await team.create(replace=True)
created_objects.append(team)
await conversation.create()
conversation_created = True
response = await team.run(
prompt="What is the capital of France? Answer in one sentence.",
params={"conversation_id": conversation.conversation_id},
)
if response is None or (
isinstance(response, str) and response.startswith("Task failed:")
):
raise RuntimeError(f"Supervised team run failed: {response}")
print("Team response:", response)
fetched = await AsyncTeam.fetch(team.team_name)
print("Supervisor agent:", fetched.attributes.supervisor_agent)
print("Supervisor task:", fetched.attributes.supervisor_task)
print("Team description:", await team.describe_team())
print("Team tools:", await team.list_tools())
finally:
if conversation_created:
await conversation.delete(force=True)
for obj in reversed(created_objects):
await obj.delete(force=True)
asyncio.run(main())
Example output (generated names, model responses, and JSON metadata vary):
output:
Team response: <answer from the supervised team>
Supervisor agent: SAMPLE_SUPERVISOR_<generated suffix>
Supervisor task: <generated supervisor task>
Team description: <JSON team metadata>
Team tools: <JSON tool metadata>
4.3. Run Team¶
AsyncTeam.run(...) starts the team workflow. The prompt argument is
passed to the task and can be referenced by task instructions using
{query}. params can include conversation_id to associate multiple
runs with the same conversation and variables to pass additional key-value
inputs.
result = await team.run(
prompt="Could you list the movies in the database?",
params={
"conversation_id": conversation_id,
"variables": {"audience": "analyst"},
},
)
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.4. Export and Import Team¶
Select AI agent teams can be exported into a portable specification and imported into the same database, a different database, or another Select AI service. The specification describes the team composition and the associated agent, task, and tool definitions that are needed to recreate the team.
AsyncTeam.export_team() returns the specification as a JSON string by
default. AsyncTeam.import_team() accepts either that JSON string or a Python
mapping containing the same team definition structure. In most cases, pass a
dict, for example the result of json.loads(exported_spec). Other
JSON-serializable collections.abc.Mapping
objects, such as OrderedDict, can also be used. On import,
profile_name identifies the Select AI profile to use in the target
database. team_name can be provided to create the imported team under a new
name; this is useful when importing into the same database as the source team.
If imported object names conflict with existing agents, tasks, tools, or teams,
set force=True to let the database replace the conflicting objects. Use this
carefully when importing into a shared schema because conflicting components can
be dropped and recreated.
import asyncio
import json
import os
import select_ai
from select_ai.agent import (
AgentAttributes,
AsyncAgent,
AsyncTask,
AsyncTeam,
TaskAttributes,
TeamAttributes,
)
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
profile_name = os.getenv("SELECT_AI_PROFILE_NAME", "LLAMA_4_MAVERICK")
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
task = AsyncTask(
task_name="EXPORT_IMPORT_MOVIE_TASK",
description="Task used by the team export/import sample",
attributes=TaskAttributes(
instruction="Help the user with movie questions. Question: {query}",
tools=[],
enable_human_tool=False,
),
)
await task.create(replace=True)
agent = AsyncAgent(
agent_name="EXPORT_IMPORT_MOVIE_ANALYST",
description="Agent used by the team export/import sample",
attributes=AgentAttributes(
profile_name=profile_name,
role="You are an AI Movie Analyst.",
enable_human_tool=False,
),
)
await agent.create(enabled=True, replace=True)
source_team = AsyncTeam(
team_name="EXPORT_IMPORT_MOVIE_TEAM",
attributes=TeamAttributes(
agents=[
{
"name": agent.agent_name,
"task": task.task_name,
}
],
process="sequential",
),
)
await source_team.create(enabled=True, replace=True)
specification = json.loads(await source_team.export())
print("Exported specification:")
print(json.dumps(specification, indent=2))
specification["name"] = "IMPORTED_MOVIE_ANALYST"
specification["task"]["task_name"] = "IMPORTED_ANALYZE_MOVIE_TASK"
await AsyncTeam.import_team(
profile_name=profile_name,
team_name="IMPORTED_MOVIE_AGENT_TEAM",
specification=specification,
force=True,
)
team = await AsyncTeam.fetch("IMPORTED_MOVIE_AGENT_TEAM")
print("Imported team:", team)
asyncio.run(main())
output:
Exported specification:
{
"name": "EXPORT_IMPORT_MOVIE_ANALYST",
"component_type": "Agent",
"task": {
"task_name": "EXPORT_IMPORT_MOVIE_TASK",
"instruction": "Help the user with movie questions. Question: {query}",
"task_attributes": {
"enable_human_tool": "false",
"tools": []
}
},
"llm_config": {
"name": "LLAMA_4_MAVERICK",
"component_type": "oci"
}
}
Imported team: AsyncTeam(team_name=IMPORTED_MOVIE_AGENT_TEAM, ...)
The same APIs can also read from or write to object storage by passing both
object_storage_credential_name and location. When exporting to object
storage, AsyncTeam.export_team() writes the specification to the location
and returns None. When importing from object storage, pass the same
credential and location instead of specification.
4.5. Inspect a team¶
AsyncTeam.describe_team() returns JSON metadata and the aggregated skills
for a team. AsyncTeam.list_tools() returns JSON metadata for the tools
available to that team.
team = await AsyncTeam.fetch("MOVIE_AGENT_TEAM")
print(await team.describe_team())
print(await team.list_tools())
The asynchronous supervised-team sample also demonstrates both inspection methods.
4.6. Lifecycle helpers¶
All async agent object types support list, fetch, enable, disable, and delete operations.
async for tool in select_ai.agent.AsyncTool.list():
print(tool.tool_name)
task = await select_ai.agent.AsyncTask.fetch("ANALYZE_MOVIE_TASK")
agent = await select_ai.agent.AsyncAgent.fetch("MOVIE_ANALYST")
team = await select_ai.agent.AsyncTeam.fetch("MOVIE_AGENT_TEAM")
await team.disable()
await team.enable()
await team.delete(force=True)
4.6.1. Object definitions¶
Use async_get_definition(object_type, object_name) to asynchronously
retrieve the canonical PL/SQL block for recreating an AI AGENT, TASK,
TOOL, or TEAM. The function returns None when the database does not
return a definition.
from select_ai.agent import async_get_definition
definition = await async_get_definition("TASK", "ANALYZE_MOVIE_TASK")
print(definition)
See the asynchronous definition sample for a complete create, inspect, and cleanup flow.
import asyncio
import os
import uuid
import select_ai
from select_ai.agent import AsyncTask, TaskAttributes, async_get_definition
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=f"SAMPLE_DEFINITION_TASK_{uuid.uuid4().hex.upper()}",
attributes=TaskAttributes(
instruction="Answer the user's question: {query}", tools=[]
),
)
await task.create(replace=True)
try:
print(await async_get_definition("TASK", task.task_name))
finally:
await task.delete(force=True)
asyncio.run(main())
The task name and exact PL/SQL formatting vary between runs and database versions. Representative output is:
output:
BEGIN
DBMS_CLOUD_AI_AGENT.CREATE_TASK(...);
END;
/
5. Async agent history¶
AsyncTeamHistory, AsyncTaskHistory, and
AsyncToolHistory provide asynchronous, read-only access to the current
user’s Select AI Agent history views. Their list() methods return async
iterators ordered from newest to oldest.
Use team_exec_id from a team event to scope task and tool history to the
same execution. Filters such as team_name, task_name,
agent_name, and tool_name can be used when an execution
identifier is not available. Tool input and output values
are decoded to Python objects when they contain valid JSON.
from select_ai.agent import (
AsyncTaskHistory,
AsyncTeamHistory,
AsyncToolHistory,
)
async for team_run in AsyncTeamHistory.list(
team_name="ORACLE_AI_DATABASE_AGENT",
limit=1,
):
print(team_run)
async for task_run in AsyncTaskHistory.list(
team_exec_id=team_run.team_exec_id
):
print(task_run)
async for tool_run in AsyncToolHistory.list(
team_exec_id=team_run.team_exec_id
):
print(tool_run)
The complete sample retrieves a team’s latest execution and uses its
team_exec_id to retrieve the associated task and tool history:
- class select_ai.agent.AsyncTeamHistory¶
Asynchronously read runs from
C##CLOUD$SERVICE.USER_AI_AGENT_TEAM_HISTORY.- classmethod list(team_name: str | None = None, team_exec_id: str | None = None, limit: int | None = None) AsyncGenerator[TeamHistoryEvent, None]¶
Yield team runs ordered from newest to oldest.
- class select_ai.agent.AsyncTaskHistory¶
Asynchronously read task runs from
C##CLOUD$SERVICE.USER_AI_AGENT_TASK_HISTORY.
- class select_ai.agent.AsyncToolHistory¶
Asynchronously read tool calls from
C##CLOUD$SERVICE.USER_AI_AGENT_TOOL_HISTORY.
import select_ai
from select_ai.agent import (
AsyncTaskHistory,
AsyncTeamHistory,
AsyncToolHistory,
)
user = os.getenv("SELECT_AI_USER")
password = os.getenv("SELECT_AI_PASSWORD")
dsn = os.getenv("SELECT_AI_DB_CONNECT_STRING")
team_name = "ORACLE_AI_DATABASE_AGENT"
async def main():
await select_ai.async_connect(user=user, password=password, dsn=dsn)
# Replace team_name with team_exec_id when the application has recorded it.
async for team_run in AsyncTeamHistory.list(team_name=team_name, limit=1):
pprint(team_run)
# team_exec_id scopes the remaining history to the same execution.
async for task_run in AsyncTaskHistory.list(
team_exec_id=team_run.team_exec_id
):
pprint(task_run)
async for tool_run in AsyncToolHistory.list(
team_exec_id=team_run.team_exec_id
):
pprint(tool_run)
asyncio.run(main())
History depends on an existing execution for the configured team. Representative output is:
output:
TeamHistoryEvent(team_exec_id='<team execution id>', team_name='ORACLE_AI_DATABASE_AGENT', state='<state>', ...)
TaskHistoryEvent(team_exec_id='<team execution id>', task_name='<task name>', state='<state>', ...)
ToolHistoryEvent(invocation_id=<invocation id>, team_exec_id='<team execution id>', tool_name='<tool name>', ...)
5.1. 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
6. Async AI agent examples¶
6.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.