Ever since I started using IRIS, I have wondered if we could create agents on IRIS. It seemed obvious: we have an Interoperability GUI that can trace messages, we have an underlying object database that can store SQL, Vectors and even Base64 images. We have a Python SDK that allows anyone to interface with the platform using Python, although this is where I felt limited as a Python developer to a certain degree. This was my attempt to create a Python SDK that can leverage several parts of IRIS to support development of agentic systems.
First, I set out to define the functional requirements:
- Developers should code primarily in Python
- Developers should not have to set configuration settings on Management Portal
- Developers should not need to code in ObjectScript
Luckily, the existing Python SDK does allow quite a bit of interfacing with the IRIS data platform. Let's explore how we can leverage them to manage context, register tools, observe messages and build agents.
Here's how I envision the SDK to be used:
from iris.agents import Agent, Production, Chat, Prompt
from iris.agents.tools import Calendar, Weather
from pydantic import BaseModel
class Response(BaseModel):
text: str
reasoning: str
weather_conversation = Chat('WeatherDiscussion')
molly_system = Prompt('MollySystemPrompt').build(scale='Fahrenheit')
alex_system = Prompt(name='AlexSystemPrompt',
text='You are an excellent assistant')
molly = Agent(
name='Molly',
description='General Assistant Agent',
system_prompt=molly_system,
model='gpt-5',
response_format=Response)
alex = Agent(
name='Alex',
description='General Assistant Agent',
system_prompt=alex_system,
model='gpt-5',
tools=[Calendar, Weather],
response_format=Response)
prod = Production(name='AgentSpace', agents=[molly, alex]).start()
molly("What's the weather in Boston today?",
chat=weather_conversation)
Let's start by defining the structure of an agent in the IRIS context. Every agent in IRIS is construed a Business Process, with their own Business Services. Every tool is construed as a Business Operation. Some tools come out of the box, such as one to query the IRIS database using SQL (also used for vector search) and one to call an LLM. The underlying database is used to store knowledge bases, prompts, agent configurations, production specs, user information, and logged information such as agent reasoning.
Before we dive into the Agents themselves, let's look at how Messages are handled. I converted each Pydantic BaseModel (useful for structured outputs) into an ObjectScript Message class stored in the namespace "Agents". If the developer defines a new BaseModel with an existing name, the structure overrides the previous one. These Message classes are converted back into Pydantic BaseModels in the LLM Business Operation, when it makes the call using appropriate libraries using Embedded Python.
class Message:
def __init__(self, name, model:BaseModel):
self.name = name
self.build_message(model)
def build_message(self, model:BaseModel):
model_json = model.model_json_schema()
cls_name = f'Agents.Message.{self.name}'
cls_text = f'''Class {cls_name} Extends (Ens.Request, %JSON.Adaptor)
{{
'''
for prop_name, prop_attribs in model_json['properties'].items():
cls_text += f'''Property {prop_name} As {r'%Double' if prop_attribs['type'] == 'number' else r'%String'}; \n'''
cls_text += '}'
irispy = get_connection(True)
stream = irispy.classMethodObject('%Stream.GlobalCharacter', '%New')
stream.invoke('Write', cls_text)
stream.invoke('Rewind')
errorlog = iris.IRISReference(None)
loadedlist = iris.IRISReference(None)
sc = irispy.classMethodValue(
'%SYSTEM.OBJ', 'LoadStream',
stream,
'ck',
errorlog,
loadedlist,
0,
'',
f'{cls_name}.cls',
'UTF-8'
)
if sc != 1:
raise RuntimeError(irispy.classMethodValue("%SYSTEM.Status", "GetErrorText", sc))
return 'Successful'
Once Messages are taken care of, here's how I created my Agent class:
class Agent:
def __init__(self,
name: str,
description: str | None = None,
system_prompt: Prompt | None = None,
model: str | None = None,
tools: list[Tool] | None = None,
response_format: BaseModel | None = None,
chat: Chat | None = None,
override: bool = True):
conn = get_connection()
cur = conn.cursor()
response_format = Message(response_format.__name__, response_format)
sql = '''SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.Tables
WHERE TABLE_TYPE='BASE TABLE'
AND TABLE_SCHEMA='SQLUser' '''
if 'Agent' not in pd.read_sql_query(sql, conn)['TABLE_NAME'].to_list():
sql = '''CREATE TABLE Agent (
agent_name VARCHAR(200) NOT NULL PRIMARY KEY,
description VARCHAR(4000),
system_prompt_id VARCHAR(200),
model VARCHAR(200),
tools VARCHAR(4000),
response_format VARCHAR(4000),
chat_id VARCHAR(200)
)'''
cur.execute(sql)
conn.commit()
sql = f"SELECT * FROM Agent WHERE agent_name = '{name}'"
agent_df = pd.read_sql_query(sql, conn)
if agent_df is not None and len(agent_df) > 0:
row = agent_df.iloc[0]
if not override:
self.name = row['agent_name']
self.description = row['description']
self.system_prompt = Prompt(row['system_prompt_id']) if row['system_prompt_id'] else None
self.model = row['model']
self.tools = row['tools']
self.response_format = row['response_format']
self.chat_id = row['chat_id']
return
sp_id = system_prompt.name if system_prompt else row['system_prompt_id']
chat_id = chat.id if chat else row['chat_id']
sql = f'''UPDATE Agent SET
description = '{description}',
system_prompt_id = '{sp_id}' ,
model = '{model}',
tools = '{str(tools)}',
response_format = '{response_format.name if response_format else None}',
chat_id = '{chat_id}'
WHERE agent_name = '{name}' '''
cur.execute(sql)
conn.commit()
self.name = name
self.description = description
self.system_prompt = Prompt(sp_id) if sp_id else None
self.model = model
self.tools = tools
self.response_format = response_format
self.chat = chat
return
if any(x is None for x in (description, model, response_format)):
raise KeyError("Missing required fields to create a new agent.")
sp_id = system_prompt.name if system_prompt else None
chat_id = chat.id if chat else None
sql = f'''INSERT INTO Agent
(agent_name, description, system_prompt_id, model, tools, response_format, chat_id)
VALUES
('{name}', '{description}', '{sp_id}', '{model}', '{str(tools)}', '{response_format.name if response_format else None}', '{chat_id}')'''
cur.execute(sql)
conn.commit()
self.name = name
self.description = description
self.system_prompt = Prompt(sp_id) if sp_id else None
self.model = model
self.tools = tools
self.response_format = response_format
self.chat = chat
def __repr__(self) -> str:
return f"Agent(name={self.name!r}, model={self.model!r}, system_prompt={getattr(self.system_prompt,'name',None)!r})"
def __call__(self, chat:Chat|None=None) -> str:
pass
Whenever Agents are initialized for the first time, it requires most of its parameters. Once an agent has been defined, it can be fetched with a simple Agent("Name") call, and the Agent's specs are loaded from database., or can be overridden by providing different specs.
For Prompts, I created a versioning system where prompts can be identified by their names (similar to Agents and Messages), but subsequent changes are versioned and stored, with the latest version being fetched when called. The prompt can also be "built" at runtime, which might allow users to inject details into a prompt template depending on the use case. All Prompts are persisted in tables.
class Prompt:
def __init__(self, name:str, text:str|None=None, iris_args:dict[str,str]|None=None):
conn = get_connection()
cur = conn.cursor()
sql = '''SELECT TABLE_SCHEMA, TABLE_NAME from INFORMATION_SCHEMA.Tables WHERE TABLE_TYPE = 'BASE TABLE' AND TABLE_SCHEMA = 'SQLUser' '''
if 'Prompt' not in pd.read_sql_query(sql, conn)['TABLE_NAME'].to_list():
sql = '''CREATE TABLE Prompt (
prompt_id VARCHAR(200) NOT NULL,
prompt_text VARCHAR(200) NOT NULL,
version INT NOT NULL,
PRIMARY KEY (prompt_id, version))'''
cur.execute(sql)
conn.commit()
sql = f'''SELECT * FROM Prompt WHERE prompt_id = '{name}' ORDER BY version DESC LIMIT 1'''
prompt_df = pd.read_sql_query(sql, conn)
last_text = None
version = 0
if prompt_df is not None and len(prompt_df) > 0:
name, last_text, version = prompt_df.iloc[0].tolist()
self.name = name
self.text = last_text
self.version = version
if not last_text and not text:
raise KeyError(f'No prompt text found for \'{name}\', and no \'text\' was provided.')
if text:
sql = f'''INSERT INTO Prompt (prompt_id, prompt_text, version) VALUES ('{name}', '{text}', {version + 1})'''
cur.execute(sql)
conn.commit()
self.text = text
self.version += 1
def __repr__(self) -> str:
return f'Prompt(name={self.name!r}, version={self.version}, text={self.text!r})'
def __str__(self) -> str:
return self.text or ''
def build(self, **vars) -> str:
import string
vars_req = {var for _, var, _, _ in string.Formatter().parse(self.text) if var}
missing = vars_req - vars.keys()
if missing:
raise KeyError(f'Missing variables {sorted(missing)} for the selected prompt')
return self.text.format(**vars)
Finally, the Production itself. The production creates the production configuration as well as the dispatch class needed to pass the REST calls to the correct Business Service (depending on which agent is being invoked).
class Production:
def __init__(self,
name: str,
agents: list[Agent],
tools: list[Tool] | None = None):
self.name = name
self.agents = agents
self.build_production()
self.create_dispatch()
def create_class(self, name, text):
irispy = get_connection(True)
stream = irispy.classMethodObject('%Stream.GlobalCharacter', '%New')
stream.invoke('Write', text)
stream.invoke('Rewind')
errorlog = iris.IRISReference(None)
loadedlist = iris.IRISReference(None)
sc = irispy.classMethodValue(
'%SYSTEM.OBJ', 'LoadStream',
stream,
'ck',
errorlog,
loadedlist,
0,
'',
f'{name}.cls',
'UTF-8'
)
if sc != 1:
raise RuntimeError(irispy.classMethodValue("%SYSTEM.Status", "GetErrorText", sc))
def create_gateway(self, name:str):
cls_text = f'''Class Agents.Gateway.{name}Service Extends Ens.BusinessService
{{
Method OnProcessInput(pInput As Agents.Message.Request, pOutput As Agents.Message.Response) As %Status
{{
set sc = ..SendRequestSync("{name}", pInput, .pResponse)
set pOutput = ##class(Agents.Message.Response).%New()
set pOutput = pResponse.%ConstructClone*(0)
Quit sc
}}
ClassMethod OnGetConnections(Output pArray As %String, pItem As Ens.Config.Item)
{{
Do ##super(.pArray, pItem)
Set pArray("{name}") = ""
}}
}}
'''
create_class(f'Agents.Gateway.{agent.name}Service', cls_text)
def create_process(self, name:str, response_format:str):
cls_text = f'''Class Agents.Process.{agent.name} Extends Ens.BusinessProcessBPL
{{
ClassMethod BuildChatJSON(pText As %String) As %String
{{
Set arr = ##class(%DynamicArray).%New()
Set obj = ##class(%DynamicObject).%New()
Do obj.%Set("role","user")
Do obj.%Set("content", pText)
Do arr.%Push(obj)
Quit arr.%ToJSON()
}}
/// BPL Definition
XData BPL [ XMLNamespace = "http://www.intersystems.com/bpl" ]
{{
<process language='objectscript' request='Agents.Message.Request' response='Agents.Message.{agent.response_format.name}'>
<context>
<property name='LLMResponse' type='Agents.Message.LLMResponse' instantiate='0' />
<property name='ChatJSON' type='%String' instantiate='0' />
</context>
<sequence>
<switch>
<case name='LLM' condition='1'>
<assign property="context.ChatJSON"
action="set"
languageOverride="objectscript"
value="##class(Agents.Process.{name}).BuildChatJSON(request.Message)" />
<call name='CallLLM' target='LLM' async='0'>
<request type='Agents.Message.LLMRequest' >
<assign property="callrequest.responseType" value=""Agents.Message.{response_format}"" action="set" />
<assign property="callrequest.chat" value="context.ChatJSON" action="set" />
</request>
<response type='Agents.Message.LLMResponse' >
<assign property="context.LLMResponse" value="callresponse" action="set"/>
</response>
</call>
<assign property="response.Message" value="context.LLMResponse.message" action="set"/>
</case>
<default>
<assign property="response.Message" value=""Hello"" action="set"/>
</default>
</switch>
</sequence>
</process>
}}
}}'''
create_class(f'Agents.Process.{name}', cls_text)
def build_production(self):
prod_xml = f'''<Production Name="{name}" LogGeneralTraceEvents="false">
<Description></Description>
<ActorPoolSize>1</ActorPoolSize>
'''
for agent in self.agents:
self.create_gateway(agent.name)
self.create_process(agent.name, agent.response_format.name)
prod_xml += f'<Item Name="{agent.name}Gateway" ClassName="Agents.Gateway.{agent.name}Service" PoolSize="1" Enabled="true"/>\n' + \
f'<Item Name="{agent.name}" ClassName="Agents.Process.{agent.name}" PoolSize="1" Enabled="true"/>\n'
prod_xml += '<Item Name="LLM" ClassName="Agents.Operation.LLM" PoolSize="1" Enabled="true"/>\n</Production>'
cls_text = f"""Class {name} Extends Ens.Production
{{
XData ProductionDefinition
{{
{prod_xml}
}}
}}
"""
create_class(name, cls_text)
def start(self):
irispy = get_connection(True)
sc = irispy.classMethodValue("Ens.Director", "StopProduction", 10, 1)
if sc != 1:
print(irispy.classMethodValue("%SYSTEM.Status","GetErrorText", sc))
irispy = get_connection(True)
sc = irispy.classMethodValue("Ens.Director", "StartProduction", self.name)
if sc != 1:
raise RuntimeError(irispy.classMethodValue("%SYSTEM.Status", "GetErrorText", sc))
print("Created/compiled/started:", self.name)
def create_dispatch(self):
cls_text = r'''
Class Agents.REST.Dispatch Extends %CSP.REST
{
XData UrlMap
{
<Routes>
<Route Url="/:agentName" Method="POST" Call="Agent" Cors="false" />
</Routes>
}
/// POST /csp/agents/{agentName}
ClassMethod Agent(agentName As %String) As %Status
{
Set %response.ContentType="application/json"
Set body = %request.Content.Read()
If body = "" {
Do %response.SetStatus(400)
Quit $$$OK
}
Set req = ##class(Agents.Message.Request).%New()
Do req.%JSONImport(body)
Set itemName = agentName _ "Gateway"
Set sc = ##class(Ens.Director).SendRequestSync(itemName, .req, .resp)
If sc '= 1 {
Do %response.SetStatus(500)
Quit $$$OK
}
Do %response.SetStatus(200)
Do %response.Write(resp.%ToJSON())
Quit $$$OK
}
}
'''
self.create_class('Agents.REST.Dispatch', cls_text)
The work here is far from over: I am currently exploring ways to send requests to the system using the dispatch class without needing to operate on Management Portal (currently it seems the web server is blocking my requests before it can reach the dispatch class). Once that is fixed, we need a few more elements to make this super useful:
NL2SQL Tool: This tool profiles a table, including creating descriptions, vectors and such. I already have created the algorithm to do this, but I intend on making it into a tool that can be directly called from Python to profile new tables, which can be leveraged by the LLM to create SQL statements.
SQL Business Operation: This tool would query the database and return the information. This would also be used by a higher level Vector Search and Index SDK that would query the database using SQL statements.
Passthrough: For Vector Search and NL2SQL profiles, a passthrough process would exist to serve the information to appropriate business services without involving agents.
Chat: Chat would exist as a table containing messages alongside chat_ids. A call to an Agent can be parameterized with a chat_id to dynamically query the database and construct the past conversation before making the LLM call. If no Chat is provided, the agentic flow remains standalone.
Note on why IRIS is uniquely positioned to help agentic application development
A typical agentic application development flow is messy. MCP tooling, context retrieval from a database, a vector db, observability (reasoning to inform prompt optimization) which is logged in a separate database and platforms like Langfuse, which itself uses multiple databases under the hood. IRIS offers a single platform to develop agents end to end, observe messages and traces on the Management Portal, and enable developers in ways few (if any) platforms can. I hope to publish this project on OpenExchange once finalized and packaged appropriately.
I hope you have enjoyed reading this article. If you have any questions, I'm always happy to discuss ideas, especially those that can change the world for the better!
.jpg)
