AI Assistant Example for PrestaShop
Notes#
Notes here
1. Introduction#
This AI Assistant is an artificial intelligence-based chat assistant integrated into the PrestaShop dashboard that communicates through Python backends and multiple specialized agents.
It is built using LangChain and LangGraph to orchestrate AI agents, tools, and workflows in a structured and modular way. LangChain provides the framework for LLM-based reasoning and tool usage, while LangGraph manages the interactions between multiple agents, enabling complex multi-agent workflows.
The purpose was to create a model of how AI agents can be used to automate events, and the context used here is PrestaShop employee management.
The module enables:
- Chat interaction directly in PrestaShop administration
- Automatic processing of employees (creation, activation, update)
- AI-based query and command traffic to PrestaShop interfaces
- Extensible architecture: adding new agents or tools is straightforward
- Secure configuration: API keys and secrets managed via OpenShift secrets
- Redis-based state management for conversation threads
Note: Each agent (reader, employee, orchestrator) runs in its own container and communicates over HTTP, allowing modular deployment and scaling.
2. Architecture#
/aiassistant
├── aiassistant.php # Main module (PrestaShop)
├── controllers/admin/
│ └── AdminAiAssistantController.php # Control panel controller
├── views/templates/admin/
│ └── aiassistant.tpl # Chat-UI Smarty-template
│
└── python-backend/
├── shared/ # Shared components and LangGraph
│ ├── app.py # FastAPI-application (chat ja new_employee)
│ ├── graph.py # LangGraph state machine between agents
│ ├── redis_memory.py # Redis memory initializer
│ ├── llm_setup.py # LLM model specification (Anthropic)
│ ├── tools.py # PrestaShop REST-tools
│ └── tools_api.py # Call interface between agents
│
├── mcp/
│ └── prestashop_connector.py # PrestaShop REST/WS Connector
│
├── orchestrator_agent/ # Central agent
│ ├── agent.py
│ └── app.py
│
├── reader_agent/ # Agent to fetch resources and information
│ ├── agent.py
│ └── app.py
│
└── employee_agent/ # Agent for employee management
├── agent.py
└── app.py
- Operating principle (information flow)
PrestaShop Admin UI (chat)
↓ Ajax
AdminAiAssistantController → Python-backend (/chat)
↓
LangGraph / orchestrator_agent
↓ decision:
↳ reader_agent (information search)
↳ employee_agent (handling of employees)
↓
prestashop_connector (REST → PrestaShop WebService API)
- When a new employee is created in PrestaShop, the module triggers a hook:
- hookActionObjectEmployeeAddAfter()
→ POST /new_employee → Python-backend
→ AI interprets and can react automatically
3. PHP module structure#
1. aiassistant.php
install():
-
Creates an API key and activates the PrestaShop webservice
-
Registers a new admin tab
-
Saves the automatically generated API key (AIASSISTANT_API_KEY)
getContent():
- Displays the status of the API key on the settings page
hookActionObjectEmployeeAddAfter():
-
Sends the new employee information to the Python backend
-
Uses a curl call to http://python-backend:8000/new_employee
-
Logs successes and errors
<?php
if (!defined('_PS_VERSION_')) {
exit;
}
class AiAssistant extends Module
{
public function __construct()
{
$this->name = 'aiassistant';
$this->tab = 'administration';
$this->version = '1.0.0';
$this->author = 'Tuukka';
$this->bootstrap = true;
parent::__construct();
$this->displayName = $this->l('AI Assistant');
$this->description = $this->l('Chat assistant for the admin page.');
}
public function install()
{
if (!parent::install() || !$this->registerTab() || !$this->registerHook('actionObjectEmployeeAddAfter')) {
return false;
}
// Taking the web service in use programmatically
Configuration::updateValue('PS_WEBSERVICE', 1);
// Create or reuse existing API key for backend communication
$apiKey = Configuration::get('AIASSISTANT_API_KEY');
if (!$apiKey) {
$apiKey = getenv('PRESTASHOP_KEY');
if (!$apiKey) {
$apiKey = bin2hex(random_bytes(16)); // fallback, if not found
}
// Register new Webservice key in PrestaShop
$apiAccess = new WebserviceKey();
$apiAccess->key = $apiKey;
$apiAccess->description = 'AI Assistant automated key';
$apiAccess->active = 1;
$apiAccess->save();
$permissions = [
'products' => ['GET' => 1],
'employees' => ['GET' => 1, 'POST' => 1, 'PUT' => 1],
];
WebserviceKey::setPermissionForAccount($apiAccess->id, $permissions);
// Persist key both in environment and database
putenv("PS_API_KEY=$apiKey");
Configuration::updateValue('AIASSISTANT_API_KEY', $apiKey);
}
return true;
}
public function uninstall()
{
$id_tab = Tab::getIdFromClassName('AdminAiAssistant');
if ($id_tab) {
$tab = new Tab($id_tab);
$tab->delete();
}
return parent::uninstall();
}
/**
* Registers a new Admin tab for the module under "Improve" section.
*/
private function registerTab()
{
$tab = new Tab();
$tab->class_name = 'AdminAiAssistant';
$tab->module = $this->name;
$id_parent = Tab::getIdFromClassName('IMPROVE');
if (!$id_parent) {
$id_parent = 0;
}
$tab->id_parent = $id_parent;
// Add name translations for all languages
$tab->name = [];
foreach (Language::getLanguages(false) as $lang) {
$tab->name[$lang['id_lang']] = $this->trans(
'AI Assistant',
[],
'Modules.Aiassistant.Admin'
);
}
return $tab->add();
}
/**
* Displays a small settings page showing current API key status.
* Helps in debugging connection issues between PrestaShop and Python backend.
*/
public function getContent()
{
$apiKeyEnv = getenv('PS_API_KEY');
$apiKeyDb = Configuration::get('AIASSISTANT_API_KEY');
$html = '<h2>AI Assistant - API avain</h2>';
$html .= '<p><strong>Ympäristömuuttuja (PS_API_KEY):</strong> '
. ($apiKeyEnv ? $apiKeyEnv : '<em>ei asetettu</em>')
. '</p>';
$html .= '<p><strong>Tietokannassa (AIASSISTANT_API_KEY):</strong> '
. ($apiKeyDb ? $apiKeyDb : '<em>ei asetettu</em>')
. '</p>';
if ($apiKeyDb) {
$html .= '<p style="color:green"><strong>Status:</strong> Avain on tallessa ja käytettävissä.</p>';
} else {
$html .= '<p style="color:red"><strong>Status:</strong> Avain puuttuu! Asennus ei onnistunut täysin.</p>';
}
return $html;
}
public function hookActionObjectEmployeeAddAfter($params)
{
if (!isset($params['object'])) {
Logger::addLog('AI Assistant: employee object missing in hook params', 3);
return;
}
$employee = $params['object'];
if (!($employee instanceof Employee)) {
Logger::addLog('AI Assistant: hook triggered but object is not Employee', 3);
return;
}
Logger::addLog("AI Assistant: New employee added - {$employee->firstname} {$employee->lastname}", 1);
// Prepare payload for backend
$data = [
'id' => $employee->id,
'firstname' => $employee->firstname,
'lastname' => $employee->lastname,
'email' => $employee->email
];
// Send JSON via HTTP POST to Python backend
$ch = curl_init("http://python-backend:8000/new_employee");
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_HTTPHEADER, ["Content-Type: application/json"]);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($data));
// Set timeouts for connection reliability
curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 30);
curl_setopt($ch, CURLOPT_TIMEOUT, 120);
$response = curl_exec($ch);
if ($response === false) {
Logger::addLog('AI Assistant: curl error - ' . curl_error($ch), 3);
} else {
Logger::addLog('AI Assistant: POST to backend successful - response: ' . $response, 1);
}
curl_close($ch);
}
}
2. controllers/admin/AdminAiAssistantController.php
-
Creates an admin page with the Smarty template aiassistant.tpl
-
Supports AJAX request /displayAjaxAsk, which sends the user's question to the backend
-
Returns the response as JSON directly to the chat
<?php
class AdminAiAssistantController extends ModuleAdminController
{
public function __construct()
{
$this->bootstrap = true;
parent::__construct();
}
/**
* Initializes content for the Admin panel page.
* Assigns Smarty variables and sets the template.
*/
public function initContent()
{
parent::initContent();
$this->context->smarty->assign([
'token' => $this->token,
'module_dir' => $this->module->getPathUri(),
]);
$this->setTemplate('aiassistant.tpl');
}
/**
* AJAX endpoint used by the chat UI.
*
* - Receives user's question as JSON.
* - Sends the question to Python backend `/chat`.
* - Returns the AI's response as JSON.
*/
public function displayAjaxAsk()
{
$input = json_decode(file_get_contents('php://input'), true);
$question = $input['question'] ?? '';
if (!$question) {
die(json_encode(['error' => 'No question provided']));
}
// Send chat query to Python backend
$ch = curl_init("http://python-backend:8000/chat");
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_HTTPHEADER, ["Content-Type: application/json"]);
curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode(["question" => $question]));
curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 30);
curl_setopt($ch, CURLOPT_TIMEOUT, 120);
$response = curl_exec($ch);
$err = curl_error($ch);
curl_close($ch);
if ($err) {
die(json_encode(['error' => $err]));
}
// Parse backend response and send back to JS
$response_data = json_decode($response, true);
$answer = $response_data['answer'] ?? 'No answer from backend';
header('Content-Type: application/json');
die(json_encode(['answer' => $answer]));
}
}
3. views/templates/admin/aiassistant.tpl
-
Simple chat interface:
-
User questions are sent to the PHP controller with a fetch() call
-
Answers are displayed in the chat window
<div class="panel">
<h3>{l s='AI Assistant' mod='aiassistant'}</h3>
<div id="chat-window" class="border p-2 mb-2" style="min-height: 200px; max-height: 400px; overflow-y: auto;" data-token="{$token|escape:'html'}"></div>
<input type="text" id="ai-question" placeholder="Kirjoita kysymys" class="form-control mb-2">
<button id="ai-send" class="btn btn-primary mb-2">Kysy</button>
</div>
<script type="text/javascript">
// Elements
const chatWindow = document.getElementById('chat-window');
const input = document.getElementById('ai-question');
const sendBtn = document.getElementById('ai-send');
// Get the token from the data- attribute
const chatToken = chatWindow.dataset.token;
console.log('Token:', chatToken);
// Add a message to chat window
function addMessage(sender, text) {
const msg = document.createElement('div');
msg.classList.add('mb-1');
msg.innerHTML = '<strong>' + sender + ':</strong> ' + text;
chatWindow.appendChild(msg);
chatWindow.scrollTop = chatWindow.scrollHeight;
}
// Click event
sendBtn.addEventListener('click', function() {
const question = input.value.trim();
console.log('Klikattu: question =', question);
if (!question) {
console.log('Empty question, not sended');
return;
}
addMessage('You', question);
input.value = '';
fetch('index.php?controller=AdminAiAssistant&ajax=1&action=ask&token=' + chatToken, {
method: 'POST',
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ question })
})
.then(res => {
return res.text();
})
.then(text => {
try {
const data = JSON.parse(text);
if (data.error) addMessage('Error', data.error);
else addMessage('Assistant', data.answer);
} catch(e) {
addMessage('Error', 'No JSON, received HTML: ' + text.substring(0, 200));
}
})
.catch(err => {
console.error('Error in fetch:', err);
addMessage('Error', err);
});
});
// Send with enter
input.addEventListener('keypress', function(e) {
if (e.key === 'Enter') sendBtn.click();
});
</script>
4. Python-backend#
The backend consists of a FastAPI application and a LangGraph state machine that routes traffic to different agents.
1. shared/app.py
-
Main application (FastAPI)
-
Endpoints:
-
/chat: handles chat questions
-
/new_employee: responds to new employees
-
graph_app variable is the global LangGraph application that connects agents.
from fastapi import FastAPI
from pydantic import BaseModel
import asyncio
import traceback
from langchain_core.messages import AIMessage
from shared.graph import create_graph_app
from contextlib import asynccontextmanager
graph_app = None # Global LangGraph instance
@asynccontextmanager
async def lifespan(app: FastAPI):
# Lifecycle handler: initializes LangGraph at startup.
global graph_app
print("Starting FastAPI & initializing LangGraph...", flush=True)
graph_app = await create_graph_app()
print("LangGraph initialized and ready!", flush=True)
yield
print("Application shutting down...", flush=True)
app = FastAPI(lifespan=lifespan)
# Data models for endpoints
class ChatRequest(BaseModel):
question: str
class EmployeeEvent(BaseModel):
id: int
firstname: str
lastname: str
email: str
# Endpoint: triggered by PrestaShop hook
@app.post("/new_employee")
async def new_employee(event: EmployeeEvent):
print(">>> /new_employee called", flush=True)
print(f"New employee added: {event.firstname} {event.lastname}")
asyncio.create_task(process_new_employee(event))
return {"status": "accepted"}
async def process_new_employee(event: EmployeeEvent):
# Creates an internal LangGraph request to validate/activate the employee.
global graph_app
message = f"""
New employee added: {event.firstname} {event.lastname}.
Check that the employee is active. If not, change it to active.
"""
try:
print(">>> /new_employee before ainvoke", flush=True)
result = await graph_app.ainvoke(
{"messages": [{"role": "user", "content": message}]},
{"configurable": {"thread_id": "prestashop_session"}}
)
print(">>> /new_employee after ainvoke", flush=True)
# Extract AI response message
ai_messages = [
m for m in result["messages"]
if isinstance(m, AIMessage)
and isinstance(m.content, str)
and m.content.strip()
]
answer = ai_messages[-1].content if ai_messages else "No response available"
print("LangGraph answer:", answer)
except Exception as e:
print("Error in process_new_employee:", str(e), flush=True)
traceback.print_exc()
# Endpoint: chat communication from Admin panel
@app.post("/chat")
async def chat(req: ChatRequest):
global graph_app
print(">>> /chat called", flush=True)
question = req.question.strip()
if not question:
return {"error": "No question provided"}
try:
print("Got question:", question, flush=True)
result = await graph_app.ainvoke(
{"messages": [{"role": "user", "content": question}]},
{"configurable": {"thread_id": "prestashop_session"}}
)
# Extract the latest AI message
ai_messages = [
m for m in result["messages"]
if isinstance(m, AIMessage)
and isinstance(m.content, str)
and m.content.strip()
]
answer = ai_messages[-1].content if ai_messages else "No response available"
print("LangGraph answer:", answer)
return {"answer": answer}
except Exception as e:
print("Chat endpoint error:", e, flush=True)
traceback.print_exc()
return {"error": str(e)}
2. shared/graph.py
-
Defines the state machine for agents (orchestrator → reader/employee)
-
Creates a StateGraph-based workflow
-
Directs the message to the correct agent based on the subject:
-
"get_resource" → reader-agent
-
"create_employee", "update_employee" → employee-agent
from langgraph.graph import StateGraph, END, MessagesState
import httpx
from shared.redis_memory import get_redis_saver
import traceback
# URLs for each agent microservice
AGENT_ENDPOINTS = {
"reader": "http://reader-agent:8002/act",
"employee": "http://employee-agent:8003/act",
"orchestrator": "http://orchestrator-agent:8004/act",
}
async def invoke_agent(state, agent_name):
# Helper function to send a state (messages) to a given agent endpoint.
async with httpx.AsyncClient(timeout=120.0) as client:
# Normalize messages into API-friendly format
messages_payload = [
{"role": m.type if hasattr(m, "type") else m["role"],
"content": m.content if hasattr(m, "content") else m["content"]}
for m in state["messages"]
]
payload = {
"messages": messages_payload,
"thread_id": "prestashop_session"
}
print(f"Sending request to {agent_name}: {AGENT_ENDPOINTS[agent_name]}")
try:
r = await client.post(AGENT_ENDPOINTS[agent_name], json=payload)
r.raise_for_status()
return {"messages": [{"role": "assistant", "content": r.json().get("answer", "")}]}
except Exception as e:
print(f"Error contacting {agent_name}: {e}")
print("invoke_agent error:", e, flush=True)
traceback.print_exc()
raise
async def create_graph_app():
# Creates the LangGraph workflow that connects orchestrator → reader/employee agents.
workflow = StateGraph(state_schema=MessagesState)
# Define agent nodes
async def orchestrator_node(state):
return await invoke_agent(state, "orchestrator")
async def reader_node(state):
return await invoke_agent(state, "reader")
async def employee_node(state):
return await invoke_agent(state, "employee")
# Register agent nodes
workflow.add_node("orchestrator", orchestrator_node)
workflow.add_node("reader", reader_node)
workflow.add_node("employee", employee_node)
# Routing logic: decides next step based on orchestrator output
def route_orchestrator(state):
last = state["messages"][-1]
#last_message = getattr(last, "content", None) or last.get("content")
if isinstance(last, dict):
last_message = last.get("content")
else:
last_message = getattr(last, "content", None)
if "get_resource" in last_message:
return "reader"
elif "create_employee" in last_message or "update_employee" in last_message:
return "employee"
return END
# Register graph transitions
workflow.add_conditional_edges("orchestrator", route_orchestrator)
workflow.add_edge("reader", END)
workflow.add_edge("employee", END)
workflow.set_entry_point("orchestrator")
# Connect Redis checkpointing for session persistence
ctx = await get_redis_saver()
async with ctx as redis_checkpointer:
graph_app = workflow.compile(checkpointer=redis_checkpointer)
print("Graph compiled successfully", flush=True)
return graph_app
3. shared/llm_setup.py
-
Initializes the Anthropic Claude model (claude-opus-4-1-20250805)
-
Uses the environment variable ANTHROPIC_API_KEY
from langchain_anthropic import ChatAnthropic
import os
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
llm = ChatAnthropic(
model="claude-opus-4-1-20250805",
api_key=ANTHROPIC_API_KEY
)
4. shared/redis_memory.py
-
Connects to Redis and provides memory management
-
Supports automatic retries if Redis is not ready immediately
import os
import asyncio
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
async def get_redis_saver():
"""
Attempts to connect to Redis multiple times with delay.
Returns an AsyncRedisSaver object when connection succeeds.
Raises:
RuntimeError: if Redis cannot be reached after all retries.
"""
redis_url = os.getenv("REDIS_URL", "redis://redis:6379")
max_retries = 10
delay = 2
for attempt in range(1, max_retries + 1):
try:
print(f"Attempting Redis connection ({attempt}/{max_retries}) at {redis_url}", flush=True)
saver = AsyncRedisSaver.from_conn_string(redis_url)
print("Using Redis AsyncRedisSaver", flush=True)
return saver
except Exception as e:
print(f"Redis not ready ({e}), retrying in {delay}s...", flush=True)
await asyncio.sleep(delay)
raise RuntimeError(f"Could not initialize Redis after {max_retries} attempts")
5. shared/tools.py & tools_api.py
-
Provide HTTP/REST-based functions for communicating with PrestaShop
-
Use the prestashop_connector.py microservice
# shared/tools_api.py
import httpx
# Base URLs for each agent
AGENT_URLS = {
"reader": "http://reader-agent:8002/act",
"employee": "http://employee-agent:8003/act",
"orchestrator": "http://orchestrator-agent:8004/act",
}
async def call_agent(agent: str, messages, thread_id: str):
"""
Asynchronously calls another AI agent via its /act endpoint.
Args:
agent (str): agent key (reader | employee | orchestrator)
messages (list): message history array
thread_id (str): session identifier
Returns:
str: textual answer returned by the called agent
"""
url = AGENT_URLS[agent]
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.post(url, json={"messages": messages, "thread_id": thread_id})
resp.raise_for_status()
return resp.json().get("answer", "")
# shared/tools.py
import requests
# Fetches a PrestaShop resource (e.g. employees, products) via REST API.
def get_resource(resource: str):
url = f"http://prestashop-connector:8001/query/{resource}"
r = requests.get(url)
r.raise_for_status()
return r.json()
# Creates a new employee record in PrestaShop.
def create_employee(employee: dict):
url = "http://prestashop-connector:8001/query/create_employee"
r = requests.post(url, json=employee)
r.raise_for_status()
return r.json()
Updates an existing employee record in PrestaShop.
def update_employee(employee_id: int, employee: dict):
url = f"http://prestashop-connector:8001/query/update_employee/{employee_id}"
r = requests.put(url, json=employee)
r.raise_for_status()
return r.json()
5. agents#
1. Orchestrator-agent
-
Acts as a decision-making layer at the beginning of LangGraph.
-
Chooses whether to call reader_agent or employee_agent.
-
Uses tools:
-
call_reader_agent
-
call_employee_agent
-
Implemented with create_react_agent construct.
# orchestrator_agent/agent.py
from shared.llm_setup import llm
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import StructuredTool
from shared.tools_api import call_agent
description = """
You are the orchestrating agent. Decide which other agent will handle the user's request:
- Reader-agent: only retrieves information from prestashop resources, such as employees or products.
- Employee-agent: can add new employees or modify existing employees.
"""
# Define a tool that lets the orchestrator call the Reader Agent.
reader_tool = StructuredTool.from_function(
coroutine=lambda messages, thread_id: call_agent("reader", messages, thread_id),
name="call_reader_agent",
description="Calls the reader agent to retrieve information from PrestaShop resources."
)
# Define a tool that lets the orchestrator call the Employee Agent.
employee_tool = StructuredTool.from_function(
coroutine=lambda messages, thread_id: call_agent("employee", messages, thread_id),
name="call_employee_agent",
description="Calls the employee agent to process employees."
)
# Create the Orchestrator ReAct Agent.
# LangGraph's `create_react_agent` allows the model to reason ("think") and act
orchestrator_react_agent = create_react_agent(
model=llm,
tools=[reader_tool, employee_tool],
prompt=description,
)
# orchestrator_agent.py/app.py
from fastapi import FastAPI, Request
from pydantic import BaseModel, field_validator, ValidationError
from typing import List, Dict, Union
from langchain_core.messages import AIMessage
from orchestrator_agent.agent import orchestrator_react_agent
app = FastAPI()
# Data model for validating incoming requests to the /act endpoint.
class AgentRequest(BaseModel):
messages: Union[List[Dict], List[str], str]
thread_id: str
# Normalize incoming messages into a consistent list of {role, content}
# objects. This ensures compatibility with LangChain message formats.
@field_validator("messages", mode="before")
def normalize_messages(cls, v):
if isinstance(v, dict):
return [v]
if isinstance(v, str):
return [{"role": "user", "content": v}]
if isinstance(v, list):
if all(isinstance(x, str) for x in v):
return [{"role": "user", "content": x} for x in v]
return v
# POST /act — Main entrypoint for AI decision flow.
# This endpoint receives chat messages from the PHP module and passes them
# to the Orchestrator Agent for reasoning.
@app.post("/act")
async def act(req: Request):
try:
data = await req.json()
print("Received JSON:", data, flush=True)
agent_req = AgentRequest(**data)
except ValidationError as ve:
# Input schema error (e.g. missing thread_id or invalid message format)
print("Validation error:", ve.json(), flush=True)
return {"error": "Invalid request format", "details": ve.errors()}
except Exception as e:
# General failure to parse the incoming JSON
print("Failed to parse JSON:", e, flush=True)
return {"error": "Failed to parse JSON", "details": str(e)}
try:
# Call the orchestrator agent asynchronously
result = await orchestrator_react_agent.ainvoke(
{"messages": agent_req.messages},
{"configurable": {"thread_id": agent_req.thread_id}}
)
# Extract AI-generated messages from the response
ai_messages = [m for m in result["messages"] if isinstance(m, AIMessage)]
answer = ai_messages[-1].content if ai_messages else "No answer"
return {"answer": answer}
except Exception as e:
# This handles runtime issues (e.g. LLM call failure, network errors)
print("Agent invocation error:", e, flush=True)
return {"error": "Agent invocation failed", "details": str(e)}
2. Reader-agent
-
Retrieves PrestaShop resources (e.g. employees, products)
-
Uses the tool:
-
get_resource(resource: str)
-
Returns data in JSON format.
# reader_agent/agent.py
from shared.tools import get_resource
from shared.llm_setup import llm
import json
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import StructuredTool
RESOURCES = [
"employees"
]
# The system prompt gives the LLM clear instructions about its role and rules.
description = f"""
You are a contributor who can retrieve information from PrestaShop.
You have the following tool at your disposal:
- get_resource: Retrieves information from PrestaShop based on the name of the resource
Use the format:
{{"tool": "get_resource", "resource": "employees"}}
The following resources are available:
{json.dumps(RESOURCES, indent=2, ensure_ascii=False)}
Rules:
- Always use **one tool at a time**.
"""
# The core tool for this agent — performs HTTP requests to PrestaShop's API
# and returns the results as JSON. The LLM can call it when it needs data.
get_resource_tool = StructuredTool.from_function(
func=get_resource,
name="get_resource",
description="Retrieves information from PrestaShop resources, e.g. employees."
)
# The agent only has one tool (get_resource) and cannot modify data.
reader_react_agent = create_react_agent(
model=llm,
tools=[get_resource_tool],
prompt=description,
)
3. Employee-agent
-
Creates and updates employees.
-
Uses tools:
-
call_reader_agent (checks for existence before creating)
-
create_employee
-
update_employee
-
Knows all PrestaShop employee fields and their requirements.
# employee_agent/agent.py
from shared.tools import create_employee, update_employee
from shared.llm_setup import llm
import json
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import StructuredTool
from shared.tools_api import call_agent
EMPLOYEE_FIELDS = {
"id_lang": {"type": "int", "required": True},
"last_passwd_gen": {"type": "string", "required": False},
"stats_date_from": {"type": "date", "required": False},
"stats_date_to": {"type": "date", "required": False},
"stats_compare_from": {"type": "date", "required": False},
"stats_compare_to": {"type": "date", "required": False},
"passwd": {"type": "string", "required": True, "max_length": 255},
"lastname": {"type": "string", "required": True, "max_length": 255},
"firstname": {"type": "string", "required": True, "max_length": 255},
"email": {"type": "string", "required": True, "max_length": 255},
"active": {"type": "bool", "required": False},
"id_profile": {"type": "int", "required": True},
"bo_color": {"type": "string", "required": False, "max_length": 32},
"default_tab": {"type": "int", "required": False},
"bo_theme": {"type": "string", "required": False, "max_length": 32},
"bo_css": {"type": "string", "required": False, "max_length": 64},
"bo_width": {"type": "int", "required": False},
"bo_menu": {"type": "bool", "required": False},
"stats_compare_option": {"type": "int", "required": False},
"preselect_date_range": {"type": "string", "required": False, "max_length": 32},
"id_last_order": {"type": "int", "required": False},
"id_last_customer_message": {"type": "int", "required": False},
"id_last_customer": {"type": "int", "required": False},
"reset_password_token": {"type": "string", "required": False, "max_length": 40},
"reset_password_validity": {"type": "date", "required": False, "nullable": True},
"has_enabled_gravatar": {"type": "bool", "required": False},
}
# Description contains extensive instructions to guide LLM behavior.
description = f"""
You are the PrestaShop employee management assistant.
You have the following tools at your disposal:
1. call_reader_agent:
Use this tool to retrieve information from the PrestaShop system, such as employees, products or other resources.
- This tool is used whenever you need to **check existing employees** before creating or editing them.
- For example, if a user requests to change the status of an employee by name,
use the `call_reader_agent` tool first to retrieve the list of employees, find the correct employee by name or email,
and then use the `update_employee` tool using the found employee ID.
- This tool returns the responses from other agents as text, which can contain JSON data about employees.
Use the format for example:
{{
"tool": "call_reader_agent",
"OPTIONS": {{
"resource": "employees"
}}
}}
2. create_employee: add an employee in JSON format to PrestaShop
Rules:
- If the user requests to create a new employee, **first search for employees with the list_employees tool.**
- If the employee already exists (same email), **do not use create_employee.**
- If there is no employee, use create_employee only once.
- Never create multiple employees with the same email.
- Only add mandatory information if no other information is provided.
- If mandatory information is missing, ask the user for it first.
Use the format for example:
{{
"tool": "create_employee",
"OPTIONS": {{
"firstname": "Matti",
"lastname": "Meikäläinen",
"email": "matti@example.com",
"passwd": "salainen123",
"id_profile": 1,
"id_lang": 1
}}
}}
3. update_employee: update employee information based on ID
Rules:
- If requested for changes based on employee name, first retrieve employees, look for ID in name, and append ID
- Only add information that is provided.
Use a format like:
{{
"tool": "update_employee",
"id": "<employee-ID>"
"OPTIONS": {{
"email": "<modified email>",
}}
}}
Remember: use the `call_reader_agent` tool whenever you need information about employees (e.g. ID, name, or email) before performing updates or creations.
The following fields are available in employee management:
{json.dumps(EMPLOYEE_FIELDS, indent=2, ensure_ascii=False)}
"""
# Allows this agent to invoke the Reader Agent when it needs to verify
# existing data before performing writes (e.g. check if email exists).
call_reader_agent = StructuredTool.from_function(
coroutine=lambda messages, thread_id: call_agent("reader", messages, thread_id),
name="call_reader_agent",
description="Calls the reader agent to retrieve information from PrestaShop resources."
)
# Adds a new employee record to PrestaShop.
create_employee_tool = StructuredTool.from_function(
func=create_employee,
name="create_employee",
description="Create a new employee in PrestaShop using JSON."
)
# Updates an existing employee’s data based on their ID.
update_employee_tool = StructuredTool.from_function(
func=update_employee,
name="update_employee",
description="Edit an existing employee in PrestaShop in JSON format."
)
# Combine tools into the Employee ReAct Agent.
# The ReAct agent uses reasoning to decide which tool to call and when.
employee_react_agent = create_react_agent(
model=llm,
tools=[create_employee_tool, update_employee_tool, call_reader_agent],
prompt=description,
)
6. Installation#
1. Secrets
-
Create secret on the server for each environment variable
-
Create a file aiassistant-secrets.env
# - PRESTASHOP_URL must include /api at the end
# - PRESTASHOP_KEY should match the key configured in PrestaShop admin panel (NOT random)
# - ANTHROPIC_API_KEY is required for LLM (Claude / Anthropic)
[rocky@rocky-vm ~]$ cat aiassistant-secrets.env
PRESTASHOP_URL=<PRESTASHOP-API-URL>
PRESTASHOP_KEY=<PRESTASHOP-API-KEY>
ANTHROPIC_API_KEY=<AI-API-KEY>
- Load file
- Docker registry secret to pull private images from GitLab
oc create secret docker-registry gitlab-registry-secret \
--docker-server=gitlab.labranet.jamk.fi:4567 \
--docker-username=<gitlab-username> \
--docker-password=<personal-access-token> \
--docker-email=<email>
2. Building & publishing container images
Each microservice (Python backend and all LangGraph agents) is built and published
automatically to GitLab’s Container Registry using Kaniko inside .gitlab-ci.yml.
All Dockerfiles are located under modules/aiassistant/python-backend/:
| Service | Dockerfile path | Image tag (pushed) |
|---|---|---|
| Python backend | shared/Dockerfile |
$CI_REGISTRY_IMAGE/python-backend:rahti-latest |
| PrestaShop connector | mcp/Dockerfile |
$CI_REGISTRY_IMAGE/prestashop-connector:rahti-latest |
| Reader agent | reader_agent/Dockerfile |
$CI_REGISTRY_IMAGE/reader-agent:rahti-latest |
| Employee agent | employee_agent/Dockerfile |
$CI_REGISTRY_IMAGE/employee-agent:rahti-latest |
| Orchestrator agent | orchestrator_agent/Dockerfile |
$CI_REGISTRY_IMAGE/orchestrator-agent:rahti-latest |
- Example Dockerfile (Python backend)
FROM python:3.11-slim
WORKDIR /app
COPY shared/requirements.txt ./
RUN pip install --upgrade pip && pip install --no-cache-dir -r requirements.txt
COPY shared ./shared
CMD ["uvicorn", "shared.app:app", "--host", "0.0.0.0", "--port", "8000"]
- Example requirements.txt
fastapi
uvicorn
requests
openai
anthropic
langgraph
langchain
langchain-anthropic
langchain-core
redis
langgraph-checkpoint-redis
langchain-openai
langchain-community
langchain-litellm
- Example .gitlab-ci.yml (build stage)
stages:
- build
build_python_backend:
stage: build
image:
name: gcr.io/kaniko-project/executor:debug
entrypoint: [""]
script:
# Authenticate GitLab registry + DockerHub (for base images)
- echo "{\"auths\":{\"$CI_REGISTRY\":{\"username\":\"$CI_REGISTRY_USER\",\"password\":\"$CI_REGISTRY_PASSWORD\"},\"https://index.docker.io/v1/\":{\"username\":\"$DOCKER_USER\",\"password\":\"$DOCKER_PAT\"}}}" > /kaniko/.docker/config.json
# Build and push Python backend image
- /kaniko/executor --context $CI_PROJECT_DIR/modules/aiassistant/python-backend \
--dockerfile $CI_PROJECT_DIR/modules/aiassistant/python-backend/shared/Dockerfile \
--destination $CI_REGISTRY_IMAGE/python-backend:rahti-latest
only:
- rahti
tags:
- general
(Other services — reader-agent, employee-agent, orchestrator-agent, etc. — follow the same pattern with their respective Dockerfiles.)
3. Deploy AI-assistant module
- This will deploy the AI assistant module for Rahti OpenShift, which PrestaShop has previously deployed to
# -------------------------------
# AI Assistant microservices stack
# Deployment order:
# 1. Redis (shared cache)
# 2. prestashop-connector
# 3. python-backend (shared tools / API entry)
# 4. reader-agent
# 5. employee-agent
# 6. orchestrator-agent
# -------------------------------
# Each agent listens on its own port and communicates via HTTP internally.
# All images should come from GitLab registry and be tagged (avoid `latest` in production).
# -------------------------------
apiVersion: v1
kind: Service
metadata:
name: redis
spec:
selector:
app: redis
ports:
- protocol: TCP
port: 6379
targetPort: 6379
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: redis
spec:
replicas: 1
selector:
matchLabels:
app: redis
template:
metadata:
labels:
app: redis
spec:
containers:
- name: redis
image: redis/redis-stack-server:latest
ports:
- containerPort: 6379
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
# NOTE: emptyDir is ephemeral — all data lost on pod restart
# For persistent cache or logs, replace with:
# persistentVolumeClaim:
# claimName: redis-pvc
volumeMounts:
- name: redis-data
mountPath: /data
volumes:
- name: redis-data
emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
name: prestashop-connector
spec:
selector:
app: prestashop-connector
ports:
- protocol: TCP
port: 8001
targetPort: 8001
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: prestashop-connector
spec:
replicas: 1
selector:
matchLabels:
app: prestashop-connector
template:
metadata:
labels:
app: prestashop-connector
spec:
containers:
- name: prestashop-connector
image: <PRESTASHOP-CONNECTOR-IMAGE>
imagePullPolicy: Always
ports:
- containerPort: 8001
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
env:
- name: PRESTASHOP_URL
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: PRESTASHOP_URL
- name: PRESTASHOP_KEY
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: PRESTASHOP_KEY
imagePullSecrets:
- name: gitlab-registry-secret
---
apiVersion: v1
kind: Service
metadata:
name: python-backend
spec:
selector:
app: python-backend
ports:
- protocol: TCP
port: 8000
targetPort: 8000
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: python-backend
spec:
replicas: 1
selector:
matchLabels:
app: python-backend
template:
metadata:
labels:
app: python-backend
spec:
containers:
- name: python-backend
image: <PYTHON-BACKEND-IMAGE> # This is an image built from the shared folder in Python-backend folder
imagePullPolicy: Always
ports:
- containerPort: 8000
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
env:
- name: ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: ANTHROPIC_API_KEY
- name: REDIS_URL
value: redis://redis:6379
imagePullSecrets:
- name: gitlab-registry-secret
---
apiVersion: v1
kind: Service
metadata:
name: reader-agent
spec:
selector:
app: reader-agent
ports:
- protocol: TCP
port: 8002
targetPort: 8002
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: reader-agent
spec:
replicas: 1
selector:
matchLabels:
app: reader-agent
template:
metadata:
labels:
app: reader-agent
spec:
containers:
- name: reader-agent
image: <READER-AGENT-IMAGE>
imagePullPolicy: Always
ports:
- containerPort: 8002
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
env:
- name: ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: ANTHROPIC_API_KEY
- name: PRESTASHOP_URL
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: PRESTASHOP_URL
imagePullSecrets:
- name: gitlab-registry-secret
---
apiVersion: v1
kind: Service
metadata:
name: employee-agent
spec:
selector:
app: employee-agent
ports:
- protocol: TCP
port: 8003
targetPort: 8003
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: employee-agent
spec:
replicas: 1
selector:
matchLabels:
app: employee-agent
template:
metadata:
labels:
app: employee-agent
spec:
containers:
- name: employee-agent
image: gitlab.labranet.jamk.fi:4567/presta-test/ref-product-presta-shop-code-v1/employee-agent:rahti-latest
imagePullPolicy: Always
ports:
- containerPort: 8003
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
env:
- name: ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: ANTHROPIC_API_KEY
- name: PRESTASHOP_URL
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: PRESTASHOP_URL
imagePullSecrets:
- name: gitlab-registry-secret
---
apiVersion: v1
kind: Service
metadata:
name: orchestrator-agent
spec:
selector:
app: orchestrator-agent
ports:
- protocol: TCP
port: 8004
targetPort: 8004
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: orchestrator-agent
spec:
replicas: 1
selector:
matchLabels:
app: orchestrator-agent
template:
metadata:
labels:
app: orchestrator-agent
spec:
containers:
- name: orchestrator-agent
image: gitlab.labranet.jamk.fi:4567/presta-test/ref-product-presta-shop-code-v1/orchestrator-agent:rahti-latest
imagePullPolicy: Always
ports:
- containerPort: 8004
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
seccompProfile:
type: RuntimeDefault
env:
- name: ANTHROPIC_API_KEY
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: ANTHROPIC_API_KEY
- name: PRESTASHOP_URL
valueFrom:
secretKeyRef:
name: aiassistant-secret
key: PRESTASHOP_URL
imagePullSecrets:
- name: gitlab-registry-secret
Verification after deployment:
- All pods are running (
oc get pods -w) - Check logs for specific pods (
oc logs -f <pod-name>)