Agentic Workload Systems: A Tutorial for Developers
In this tutorial we will build a multi-step research agent that accepts a product category, calls mocked market-research tools, and returns a structured competitive summary. You should follow along if you need a reusable
In this tutorial we will build a multi-step research agent that accepts a product category, calls mocked market-research tools, and returns a structured competitive summary. You should follow along if you need a reusable agent loop that plans, gathers data, and synthesizes output without hidden token costs.
What you'll need
Python 3.10 or newer, the OpenAI SDK, and an API key from https://portal.oxlo.ai. Install the SDK with pip.
pip install openaiStep 1: Initialize the Oxlo.ai client
I start every agent project with a thin client wrapper. Because Oxlo.ai is fully OpenAI-compatible, this is a single import and two lines of config. I use qwen-3-32b here because Oxlo.ai lists it as a flagship for agent workflows.
from openai import OpenAI
import json
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key="YOUR_OXLO_API_KEY" # from https://portal.oxlo.ai
)
MODEL = "qwen-3-32b"Step 2: Define tools and their mock implementations
Real agents need tools. I define two: search_competitors and get_pricing_tier. The implementations return hard-coded JSON so the tutorial runs without external API keys, but the schema is identical to what you would send to a live search or CRM endpoint.
def search_competitors(category: str):
"""Return top competitors for a product category."""
database = {
"wireless earbuds": ["AudioTech Pro", "BassBoost X", "SoundWave 9"],
"cloud storage": ["NimbusDrive", "VaultSync", "DataHaven"]
}
return json.dumps({
"category": category,
"competitors": database.get(category.lower(), ["Competitor A", "Competitor B"])
})
def get_pricing_tier(competitor: str):
"""Return mock pricing for a competitor."""
prices = {
"AudioTech Pro": "$149",
"BassBoost X": "$129",
"SoundWave 9": "$199"
}
return json.dumps({
"competitor": competitor,
"price": prices.get(competitor, "N/A")
})
TOOLS = [
{
"type": "function",
"function": {
"name": "search_competitors",
"description": "Find top competitors in a category",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string"}
},
"required": ["category"]
}
}
},
{
"type": "function",
"function": {
"name": "get_pricing_tier",
"description": "Look up pricing for a named competitor",
"parameters": {
"type": "object",
"properties": {
"competitor": {"type": "string"}
},
"required": ["competitor"]
}
}
}
]Step 3: Write the agent system prompt
The system prompt is the only code that changes behavior without redeploying. I keep it strict: the agent must plan, call tools, then synthesize. No tool call means no final answer.
SYSTEM_PROMPT = """You are a market-research agent. Your job is to produce a competitive summary for a product category.
Follow this exact workflow:
1. Call search_competitors with the user-supplied category.
2. For each competitor returned, call get_pricing_tier.
3. Once you have all data, synthesize a short report with these sections: Competitors, Pricing Overview, and a one-sentence Strategic Takeaway.
4. Do not invent data. Only use values returned by the tools.
5. Return the final report as plain text with clear headers."""Step 4: Build the agent loop
This is the core pattern. I send the conversation history to Oxlo.ai, check for tool_calls, execute the local function, and append the result back to the messages list. Because Oxlo.ai charges a flat rate per request rather than per token, running a multi-turn agent loop with long context does not explode in cost the way token-based billing would. See https://oxlo.ai/pricing for current plan details.
def run_agent(user_query: str):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_query}
]
for _ in range(10): # safety cap
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
message = response.choices[0].message
assistant_msg = {
"role": message.role,
"content": message.content or ""
}
if message.tool_calls:
assistant_msg["tool_calls"] = [
{
"id": tc.id,
"type": tc.type,
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
messages.append(assistant_msg)
if message.tool_calls:
for tc in message.tool_calls:
fn_name = tc.function.name
args = json.loads(tc.function.arguments)
if fn_name == "search_competitors":
result = search_competitors(**args)
elif fn_name == "get_pricing_tier":
result = get_pricing_tier(**args)
else:
result = json.dumps({"error": "unknown tool"})
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
else:
return message.content
return "Reached max turns without final answer."Step 5: Add context trimming for long traces
For production workloads, I keep the message list from growing forever. This helper drops the earliest assistant turns once we pass ten messages, preserving the system prompt and the most recent reasoning. On Oxlo.ai, trimming context is an optimization, not a cost emergency, because each request costs the same flat rate regardless of prompt length.
def trim_messages(messages, max_len=10):
if len(messages) <= max_len:
return messages
# Preserve system prompt and the most recent exchanges
trimmed = [m for m in messages if m["role"] == "system"]
trimmed.extend(messages[-(max_len - 1):])
return trimmed
def run_agent(user_query: str):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_query}
]
for _ in range(10):
messages = trim_messages(messages)
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
message = response.choices[0].message
assistant_msg = {
"role": message.role,
"content": message.content or ""
}
if message.tool_calls:
assistant_msg["tool_calls"] = [
{
"id": tc.id,
"type": tc.type,
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
messages.append(assistant_msg)
if message.tool_calls:
for tc in message.tool_calls:
fn_name = tc.function.name
args = json.loads(tc.function.arguments)
if fn_name == "search_competitors":
result = search_competitors(**args)
elif fn_name == "get_pricing_tier":
result = get_pricing_tier(**args)
else:
result = json.dumps({"error": "unknown tool"})
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result
})
else:
return message.content
return "Reached max turns without final answer."Run it
Call the agent with a category and print the result.
if __name__ == "__main__":
report = run_agent("Research the wireless earbuds market")
print(report)Example output:
Competitors: AudioTech Pro, BassBoost X, SoundWave 9
Pricing Overview:
- AudioTech Pro: $149
- BassBoost X: $129
- SoundWave 9: $199
Strategic Takeaway: BassBoost X undercuts the field by $20, but SoundWave 9 commands a premium, suggesting a split market between budget and high-end audio.Wrap-up
You now have a working ReAct-style agent loop backed by Oxlo.ai. Two concrete next steps: wire the tool stubs to real APIs such as SerpApi or your internal CRM, and add a persistence layer with SQLite or Redis so the agent can resume long-running research jobs across sessions.
Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.