This post aims to demonstrate how to integrate the Model Context Protocol (MCP) into the example from the previous article (click here), where I used the Agent2Agent Protocol (A2A) to implement a CRUD system. Therefore, I highly recommend reading the previous post first, as it provides the foundation to better understand this content.
For better context, the use case is a CRUD system where a manager agent delegates tasks to the CRUD agent and the metrics agent. Their purpose is to create, retrieve, update, or delete data, as well as to compute user-requested metrics. The agents were built using CrewAI, LangGraph, and Hugging Face’s smolagents.
With the integration of MCP into this use case, both the agents and the tools become decoupled, which ensures easier maintenance, scalability, and system evolution. Therefore, the 2 protocols are not in conflict, on the contrary, when combined, they enable the creation of much more robust systems.
In this post, I’ll cover:
- Building the MCP Server and Client
- Creating the Agents
- Execution and Results Analysis
Below are the Python version used and the requirements.txt file.
⚠️ Python Version: 3.12.9
We’ll also use SQLite3 to store the registered products. In the previous post, I showed step by step how to create the inventory database and the products table in SQLite3. For convenience, below is the SQL command to create the products table.
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
category TEXT,
price REAL NOT NULL,
stock_qty INTEGER NOT NULL DEFAULT 0,
launch_date TEXT,
is_active INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
);
Building the MCP Server and Client
Below is the mcp_server.py file.
import asyncio
from fastmcp import FastMCP
from python_a2a import A2AClient
import sqlite3
from typing import Optional
from pandas import DataFrame
import time
last_execution_delegate_work = None
mcp_server = FastMCP("CRUDTools")
@mcp_server.tool()
async def create_product(product_name: str, category: str, price: float, stock_qty: int, launch_date: str, is_active: bool) -> str:
"""
Create a new product in the inventory database.
Args:
product_name (str): The name of the product.
category (str): The category of the product.
price (float): The price of the product.
stock_qty (int): The quantity of the product in stock.
launch_date (str): The launch date of the product. The format must be YYYY-MM-DD.
is_active (bool): Whether the product is active or not.
Returns:
A success message or an error message if the operation fails.
"""
def _create():
with sqlite3.connect("inventory.db") as con:
cur = con.cursor()
cur.execute(
"""
INSERT INTO products (name, category, price, stock_qty, launch_date, is_active)
VALUES (?, ?, ?, ?, ?, ?)
""",
(product_name, category, price, stock_qty, launch_date, is_active)
)
return f"Product '{product_name}' inserted successfully."
try:
return await asyncio.to_thread(_create)
except sqlite3.Error as e:
return f"An error occurred: {e}"
@mcp_server.tool()
async def get_products(fields: Optional[dict] = None, operator: Optional[str] = "=") -> list[dict] | str:
"""
Get products from the inventory database based on provided filters.
Args:
fields (Optional[dict]): Dictionary of filters to apply. Supported keys include:
- name (str): The name of the product.
- category (str): The category of the product.
- price (float): The price of the product.
- stock_qty (int): The stock quantity.
- launch_date (str): The launch date in format YYYY-MM-DD.
- is_active (bool): Whether the product is active.
operator (Optional[str]): The comparison operator to use for filtering. Default is "=".
Supported operators include "=", "<", ">", "<=", ">=", "!=", "LIKE".
Examples:
- get_products(fields={"category": "Electronics", "is_active": True})
- get_products(fields={"price": 100.0})
- get_products(fields={"name": "Laptop"}, operator="LIKE")
- get_products()
Returns:
A list of dictionaries with the products that match the filters or an error message if the query fails.
"""
def _get():
with sqlite3.connect("inventory.db") as con:
cur = con.cursor()
query = "SELECT * FROM products"
if fields is not None:
query += " WHERE"
for column, value in fields.items():
if operator == "LIKE":
query += f" {column.lower()} {operator} '%{value.lower()}%' AND"
else:
query += f" {column} {operator} "
if type(value) == str:
query += f"'{value}' AND"
else:
query += f"{value} AND"
query = query[:-4]
cur.execute(query)
data = cur.fetchall()
columns = [desc[0] for desc in cur.description]
df = DataFrame(data, columns=columns)
return df.to_dict(orient="records")
try:
return await asyncio.to_thread(_get)
except sqlite3.Error as e:
return f"An error occurred: {e}"
@mcp_server.tool()
async def update_product(product_id: int, fields: dict) -> str:
"""
Updates a product in the inventory database based on the provided fields.
Args:
product_id (int): The ID of the product to update.
fields (dict): The fields to update. Supported keys include:
- name (str)
- category (str)
- price (float)
- stock_qty (int)
- launch_date (str, format YYYY-MM-DD)
- is_active (bool)
Example:
update_product(1, {"name": "New Product Name", "price": 99.99})
Returns:
A success message if the update is successful, or an error message if it fails.
"""
if product_id is None and not fields:
return "Please provide a product ID and at least one field to update."
elif product_id is None:
return "Please provide a product ID to update."
elif not fields:
return "Please provide at least one field to update."
def _update():
with sqlite3.connect("inventory.db") as con:
cur = con.cursor()
data = ", ".join(f"{k} = ?" for k in fields)
values = list(fields.values()) + [product_id]
cur.execute(f"UPDATE products SET {data} WHERE id = ?", values)
return f"Product [{product_id}] updated successfully."
try:
return await asyncio.to_thread(_update)
except sqlite3.Error as e:
return f"An error occurred: {e}"
@mcp_server.tool()
async def delete_product(product_id: Optional[int] = None, product_name: Optional[str] = None) -> str:
"""
Delete a product from the inventory database.
Args:
product_id (Optional[int]): The ID of the product to delete.
product_name (Optional[str]): The name of the product to delete.
Returns:
A success message or an error message if the operation fails.
"""
if product_id is None and product_name is None:
return "Please provide either a product ID or a product name to delete."
if product_id is not None and product_name is not None:
condition = f"id = {product_id} AND name = '{product_name}'"
elif product_id is not None:
condition = f"id = {product_id}"
else:
condition = f"name = '{product_name}'"
def _delete():
with sqlite3.connect("inventory.db") as con:
cur = con.cursor()
cur.execute(f"DELETE FROM products WHERE {condition}")
return f"Product [{product_id}] deleted successfully."
try:
return await asyncio.to_thread(_delete)
except sqlite3.Error as e:
return f"An error occurred: {e}"
@mcp_server.tool()
async def delegate_work(url: str, user_input: str) -> str:
"""
Delegates work for a specific agent based on what was requested by the user
Args:
- url (str): url contained in the agent card
- user_input (str): user request
Returns:
- Agent response to task
"""
global last_execution_delegate_work
current_execution = {
"url": url,
"user_input": user_input,
"time": time.time()
}
if last_execution_delegate_work and last_execution_delegate_work["url"] == url and last_execution_delegate_work["user_input"] == user_input and (current_execution["time"] - last_execution_delegate_work["time"]) < 60:
return last_execution_delegate_work["result"]
client = A2AClient(url)
try:
result = await asyncio.to_thread(client.ask, user_input)
last_execution_delegate_work = {**current_execution, "result": result}
return result
except Exception as e:
return f"Delegate work failed: {e}"
if __name__ == "__main__":
mcp_server.run(transport="sse", host="0.0.0.0", port=4000)
You can learn more about how to create your own MCP server using FastMCP by clicking here. In that tutorial, the server is implemented synchronously. In the current use case, however, I made the tools asynchronous, since multiple calls to the MCP server within the same execution could otherwise lead to a circular dependency.
For example, imagine the user requests to insert a product. The flow would be:
User → Manager Agent → MCP Server (delegate_work) → CRUD Agent → MCP Server (create_product)
In this scenario, delegate_work delegates the task to the CRUD agent, which in turn uses the product creation tool. The issue is that delegate_work would remain blocked waiting for a response, but that response depends on create_product, which also lives in the MCP server. This creates a deadlock or timeout, as the server ends up waiting on itself, causing multiple redirects, communication failures, etc.
By making the calls asynchronous, however, the event loops can handle this circular dependency, allowing the request to complete without blocking.
In addition, I implemented a simple caching layer using the last_execution_delegate_work variable. The purpose of the caching layer is to skip repeated requests to the same agent with the same input by checking whether the URL (target agent) and input have already been processed within the last 60 seconds. If it’s the same request, the result from the first execution is returned immediately, without resending the call to the agent. This ensures that each agent is invoked only once per task, even if the manager repeats the tool call.
In addition to the MCP server, below is the mcp_client.py file, which the agents will use to consume the available tools.
from langchain_mcp_adapters.client import MultiServerMCPClient
def get_mcp_client(option: str="langchain") -> MultiServerMCPClient | list[dict]:
mcp_server_config = {
"crud-mcp-server": {
"transport": "sse",
"url": "http://localhost:4000/sse/"
}
}
if option == "langchain":
return MultiServerMCPClient(mcp_server_config)
return list(mcp_server_config.values())
Creating the Agents
Below is the CRUD agent script.
import asyncio
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
from mcp_client import get_mcp_client
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from dotenv import load_dotenv
load_dotenv()
@agent(
name="CRUD Agent",
description=(
"An agent that manages product records, allowing creation, "
"retrieval, update, and deletion of products from the "
"inventory database."
),
version="1.0.0"
)
class CRUDAgent(A2AServer):
def __init__(self):
super().__init__()
self.__client = get_mcp_client()
@skill(
name="Create Product",
description="Create a product",
tags=["create", "product"]
)
def create_product(self):
pass
@skill(
name="Get products",
description="Get products from the inventory database based on the provided fields.",
tags=["get", "products"]
)
def get_products(self):
pass
@skill(
name="Delete a product",
description="Delete a product from the inventory database.",
tags=["delete", "product"]
)
def delete_product(self):
pass
@skill(
name="Update a product",
description="Update a product in the inventory database.",
tags=["update", "product"]
)
def update_product(self):
pass
def handle_task(self, task):
message_data = task.message or {}
content = message_data.get("content", {})
self.__text = content.get("text", "") if isinstance(content, dict) else ""
resp = self.__run_agent()
task.artifacts = [{
"parts": [{"type": "text", "text": resp['messages'][-1].content}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task
def __run_agent(self):
tools = asyncio.run(self.__client.get_tools())
tools = [tool for tool in tools if tool.name != "delegate_work"]
prompt = (
"You are an intelligent agent responsible for managing products in an inventory database.\n"
"Handle requests to create, retrieve, update, or delete products in a clear and helpful way.\n"
"Always perform the requested action and respond with a direct message indicating success or failure."
)
model = ChatOpenAI(
model="gpt-4.1-mini",
temperature=0.1
)
agent = create_react_agent(
model=model,
tools=tools,
prompt=SystemMessage(content=prompt),
debug=True
)
return asyncio.run(agent.ainvoke({"messages": self.__text}))
if __name__ == "__main__":
agent = CRUDAgent()
run_server(agent, port=5001)
Note that, unlike in the previous post, where the tools were created and consumed directly within the agent’s server, here they are retrieved via the MCP server:
self.__client = get_mcp_client()
tools = asyncio.run(self.__client.get_tools())
In the A2A protocol, agents expose their capabilities through skills. Although the actual logic is implemented in the MCP via tools, these skills still need to be declared so the agent can make its capabilities visible to users, other servers, or agents that may interact with it. For this reason, 4 placeholder skills were created, each with a pass, solely to expose the CRUD operations that the agent provides.
Also note that both retrieving the tools and executing the agents are asynchronous calls, which prevents the circular dependency issue mentioned earlier. You’ll notice that the other 2 agents are also executed asynchronously.
Below is the metrics agent code, which is analogous to the CRUD agent. The main difference is that, while the CRUD agent already includes the asynchronous ainvoke method, the metrics agent does not. To address this, a custom asynchronous method was implemented, ensuring its operations are also executed asynchronously.
import asyncio
from python_a2a import A2AServer, agent, skill, run_server, TaskStatus, TaskState
from mcp_client import get_mcp_client
from smolagents import CodeAgent, OpenAIServerModel, ToolCollection
from dotenv import load_dotenv
load_dotenv()
@agent(
name="Metrics Agent",
description=(
"An agent that analyzes product data from the inventory database and returns relevant metrics, "
"such as averages, totals, counts, or performance insights based on price, stock levels, categories, "
"launch dates, and activity status."
),
version="1.0.0"
)
class MetricsAgent(A2AServer):
def __init__(self):
super().__init__()
self.__client = get_mcp_client(option="smolagents")
@skill(
name="Get products",
description="Get products from the inventory database based on the provided fields.",
tags=["get", "products"]
)
def get_products(self):
pass
def handle_task(self, task):
message_data = task.message or {}
content = message_data.get("content", {})
self.__text = content.get("text", "") if isinstance(content, dict) else ""
resp = asyncio.run(self.__run_agent())
task.artifacts = [{
"parts": [{"type": "text", "text": str(resp)}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task
async def __run_agent(self):
with ToolCollection.from_mcp(self.__client, trust_remote_code=True) as tool_collection:
tools = [*tool_collection.tools]
tools = [tool for tool in tools if tool.name == "get_products"]
prompt = (
"You are an intelligent agent responsible for generating metrics and insights based on product data from the inventory database.\n"
"When asked for any type of summary, total, average, or analytical information, respond with the correct calculation based on the available data.\n"
"Always perform the necessary computations and return concise, factual responses with the result of the analysis.\n\n"
"Tip: if the tool's output is a string, use json.loads to convert it to a list of dictionaries. After that, you can use pandas/numpy to perform the calculations."
)
model = OpenAIServerModel(
model_id="gpt-4.1-mini",
temperature=0.1
)
agent = CodeAgent(
model=model,
max_steps=10,
tools=tools,
additional_authorized_imports=["pandas", "numpy", "json"],
verbosity_level=2
)
resp = agent.run(
task=prompt,
additional_args={"user_input": self.__text}
)
return resp
if __name__ == "__main__":
agent = MetricsAgent()
run_server(agent, port=5002)
Finally, here is the manager agent code, which is analogous to both the CRUD agent and the metrics agent.
import asyncio
from python_a2a import A2AServer, A2AClient, agent, skill, TaskStatus, TaskState, run_server
from mcp_client import get_mcp_client
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter
from dotenv import load_dotenv
load_dotenv()
AGENT_SERVERS = [
"http://localhost:5001",
"http://localhost:5002"
]
@agent(
name="Manager Agent",
description="Select the most suitable agent to fulfill the user's request.",
version="1.0.0"
)
class ManagerAgent(A2AServer):
def __init__(self):
super().__init__()
self.cards = []
for url in AGENT_SERVERS:
a2a_client = A2AClient(url)
self.cards.append(a2a_client.agent_card)
self.__mcp_client = get_mcp_client(option="crewai")
@skill(
name="Delegate work",
description="Delegate work to the most suitable agent based on the user's request.",
tags=["delegate", "work"]
)
def delegate_work(self):
pass
def handle_task(self, task):
message_data = task.message or {}
content = message_data.get("content", {})
self.__text = content.get("text", "") if isinstance(content, dict) else ""
resp = asyncio.run(self.__run_manager())
task.artifacts = [{
"parts": [{"type": "text", "text": resp.raw}]
}]
task.status = TaskStatus(state=TaskState.COMPLETED)
return task
async def __run_manager(self):
with MCPServerAdapter(self.__mcp_client) as tools:
tools = [tool for tool in tools if tool.name == "delegate_work"]
try:
manager = Agent(
role="Task Orchestrator Agent",
goal=(
"Understand the user's goal and delegate the request to the most suitable agent available, "
"ensuring the task is completed efficiently and accurately."
),
backstory=(
"You are an experienced orchestrator of intelligent agents. Your expertise lies in interpreting user inputs, "
"identifying the most capable specialized agent for each task, and delegating responsibilities effectively. "
"You are logical, decisive, and focused on matching problems to the right skills within your team. "
"You never perform the task yourself—you always assign it to the best-suited agent in your team."
),
tools=tools,
llm="gpt-4.1",
verbose=True
)
except Exception as e:
print(f"Error creating manager agent: {e}")
manager_task = Task(
description=(
"User input: {user_input}\n\n"
"You are the orchestrator of a team of specialized agents. Your job is to understand the user's request, improve its clarity if needed, and delegate the execution to the right agent(s).\n\n"
"Instructions:\n"
"1. Carefully analyze the user's input and rewrite it in a clearer and more structured form for the subordinate agents.\n"
"2. If the user is requesting data retrieval, updates, or deletions:\n"
" a. Always perform a **validation step BEFORE executing the main task**.\n"
" b. Use the CRUD agent to retrieve available data (e.g., list of product names, categories, or IDs).\n"
" c. Try to match the user's input against this data — even if it's partially written, incomplete, or contains spelling variations.\n"
" d. If a clear match is found, **rewrite the user input accordingly** (e.g., change 'graphic tee urban' to 'Graphic Tee Urban Style').\n"
" e. Then, delegate the clarified request to the proper agent.\n\n"
"3. If the request is about creating/inserting a new item:\n"
" - Do NOT perform validations on whether the product already exists unless explicitly requested by the user.\n"
" - Assume the user is introducing a new product unless they say otherwise.\n"
"4. If the request involves metrics or calculations:\n"
" a. If the metric request is general (e.g., average price of all products), pass it directly to the metrics agent.\n"
" b. If the metric request contains ambiguous or fuzzy filters (e.g., 'books about Python'), first validate the referenced items using the CRUD agent.\n"
" c. Once you have confirmed the correct references (e.g., product names), **rewrite the request to include those exact values** as filters for the metrics agent.\n"
" d. Do not manually extract product data or replicate metric logic — the metrics agent will handle calculations.\n\n"
"5. If multiple agents are needed (e.g., validation + metrics), coordinate the steps.\n"
"6. Assign subtasks to the relevant agents using the available tools and skills.\n"
"7. Collect all outputs from the agents and formulate a final response that clearly and directly answers the user's request.\n\n"
"Available agents and their capabilities:\n"
f"{self.cards}\n\n"
"Important:\n"
"- You do not execute tasks yourself; you only orchestrate by delegating to other agents.\n"
"- Always rewrite the input in a more structured and helpful form before delegating.\n"
"- Always validate uncertain or fuzzy user input for GET/UPDATE/DELETE before execution.\n"
"- Your final answer must be clear and helpful based on the subordinate agents' outputs."
),
expected_output=(
"A clear and helpful response to the user's input, based on the output generated by the selected specialized agent."
),
agent=manager
)
crew = Crew(
agents=[manager],
tasks=[manager_task],
verbose=True,
process=Process.sequential
)
return crew.kickoff(inputs={"user_input": self.__text})
if __name__ == "__main__":
agent = ManagerAgent()
run_server(agent, port=5000)
Execution and Results Analysis
First, a product will be inserted into the products table. In this example, we’ll see how the manager agent and the CRUD agent interact with their respective tools through the MCP server. Since I already provided a detailed log analysis in the previous post, I won’t repeat it here for all examples. Instead, I’ll focus on showing the inputs provided and their corresponding outputs.
The product creation request was: “Can you please add Wireless Earbuds Pro to the Electronics section? It costs 149.99, we have 50 in stock, launched on 2025-08-01, and it’s active.”.
The manager agent then uses the delegate_work tool to delegate this task to the CRUD agent. This process can be seen in the manager’s log, where it makes a request to the endpoint http://localhost:4000/sse, which is the MCP server execution point. In the log snippet below, you can follow how the manager structured the input and delegated the creation to the CRUD agent.
INFO:httpx:HTTP Request: GET http://localhost:4000/sse/ "HTTP/1.1 307 Temporary Redirect"
INFO:httpx:HTTP Request: GET http://localhost:4000/sse "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=3b9ad105cfbe4c8aa42748edc63645f6 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=3b9ad105cfbe4c8aa42748edc63645f6 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=3b9ad105cfbe4c8aa42748edc63645f6 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=3b9ad105cfbe4c8aa42748edc63645f6 "HTTP/1.1 202 Accepted"
╭──────────────────────────────────────────────────────────── 🔧 Agent Tool Execution ─────────────────────────────────────────────────────────────╮
│ │
│ Agent: Task Orchestrator Agent │
│ │
│ Thought: Thought: The user wants to add (create) a new product called "Wireless Earbuds Pro" in the Electronics section with specific │
│ attributes (price, stock, launch date, status). This is a straightforward "create product" task and does not require validation or checking │
│ for existing products, as per instructions for creation tasks. My job is to clearly structure this information for the CRUD Agent and delegate │
│ the creation task. │
│ │
│ Using Tool: delegate_work │
│ │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─────────────────────────────────────────────────────────────────── Tool Input ───────────────────────────────────────────────────────────────────╮
│ │
│ "{\"url\": \"http://localhost:5001\", \"user_input\": \"Please create a new product with the following details:\\n- Name: Wireless Earbuds │
│ Pro\\n- Category: Electronics\\n- Price: 149.99\\n- Stock: 50\\n- Launch Date: 2025-08-01\\n- Status: Active\"}" │
│ │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭────────────────────────────────────────────────────────────────── Tool Output ───────────────────────────────────────────────────────────────────╮
│ │
│ The product "Wireless Earbuds Pro" has been created successfully in the inventory. │
│ │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
The CRUD agent, in turn, uses the create_product tool, also available on the MCP server, to insert the requested product into the products table. In the log below, you can see the tool being invoked with the provided parameters, followed by the confirmation that the Wireless Earbuds Pro was successfully created in the inventory.
INFO:httpx:HTTP Request: GET http://localhost:4000/sse/ "HTTP/1.1 307 Temporary Redirect"
INFO:httpx:HTTP Request: GET http://localhost:4000/sse "HTTP/1.1 200 OK"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=a0995dfecd3b4ce48ab70de23dd42984 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=a0995dfecd3b4ce48ab70de23dd42984 "HTTP/1.1 202 Accepted"
INFO:httpx:HTTP Request: POST http://localhost:4000/messages/?session_id=a0995dfecd3b4ce48ab70de23dd42984 "HTTP/1.1 202 Accepted"
tool_calls=[{'name': 'create_product', 'args': {'product_name': 'Wireless Earbuds Pro', 'category': 'Electronics', 'price': 149.99, 'stock_qty': 50, 'launch_date': '2025-08-01', 'is_active': True}, 'id': 'call_TjkOPJvS3KgbhSFlb0j0aa7G', 'type': 'tool_call'}]
{'agent': {'messages': [AIMessage(content='The product "Wireless Earbuds Pro" has been created successfully in the inventory.', ...
Therefore, the final message returned to the user was: “The product ‘Wireless Earbuds Pro’ has been created successfully in the inventory.”.
In addition to this example, 19 more products were inserted. Click here ⬇️ to expand and view all the provided inputs.
products = [
"I need to register a Classic White T-Shirt under Clothing, priced at 19.90. There are 100 units and it was launched on 2025-07-15. It's active.",
"Insert a product: The Art of War Book. It's from the Books category, price 9.99, stock 75, release date 2025-06-10, and it's active.",
"Please include Bluetooth Smartwatch X in Electronics, launched 2025-07-25. Price is 89.50, with 30 units available, and status should be active.",
"I'd like to create a new item: Men's Denim Jacket, category Clothing, 120.00 dollars, 15 in stock, launched 2025-08-10, marked active.",
"Register Digital Photography Guide under the Books category. It costs 24.00, stock is 40, launch date is 2025-05-20, and it's active.",
"Add this: Noise Cancelling Headphones, from Electronics. Launch: 2025-08-05, price: 249.90, quantity: 20, active: true.",
"We need a new product: Women's Summer Dress. Clothing category, costs 39.99, 80 units, released 2025-07-30, and it's active.",
"Create an entry for Python Programming for Beginners in Books. Price is 35.50, stock level is 60, launch 2025-06-01, active product.",
"Please log a new product: Portable Power Bank 10000mAh, Electronics, 29.90 dollars, 120 in stock, launched 2025-08-15, set as active.",
"I'd like to list a Cotton Hoodie Unisex under Clothing, launched 2025-07-05. Price: 54.99, 40 units, status: active.",
"Enter this book into the inventory: Modern Web Design Handbook. Category: Books, cost: 45.00, 25 available, launched 2025-06-18, mark active.",
"Add Wireless Keyboard & Mouse Combo as a new product in Electronics. Priced at 59.00, 90 items, release date: 2025-08-12, it's active.",
"New product alert: Slim Fit Chinos, goes into Clothing, cost is 64.90, stock 35, launch 2025-07-22, should be active.",
"Introduce Creative Writing Essentials into the Books category. Price is 27.75, stock count 55, date: 2025-05-28, active status.",
"I want to add Home Security Camera to the Electronics list. Set the price at 89.99, quantity 45, launched on 2025-08-17, and mark it as active.",
"Log Waterproof Rain Jacket in Clothing. Launch date is 2025-07-18, price: 99.50, 22 units, make sure it's active.",
"Put History of the Modern World in the Books section. 39.80 is the price, with 38 in stock, from 2025-06-06, active item.",
"Please insert 4K Action Camera under Electronics. Tag it with 119.00 price, 18 items, launch 2025-08-19, and it's active.",
"Could you add Graphic Tee Urban Style to Clothing? Set launch for 2025-08-03, priced 24.99, 70 units in stock, active = true."
]
The following tests are related to data retrieval. Feel free to run them and review the corresponding logs.
user_input = "What is the price and stock quantity of the wireless earbuds?"
Response: The "Wireless Earbuds Pro" is priced at $149.99 and has a stock quantity of 50 units.
user_input = "Can you show me all the products available in the clothing category?"
Response: Here are all products available in the 'Clothing' category from the inventory database:
1. Classic White T-Shirt
- Price: $19.90
- Stock Quantity: 100
- Launch Date: 2025-07-15
- Active: Yes
2. Men's Denim Jacket
- Price: $120.00
- Stock Quantity: 15
- Launch Date: 2025-08-10
- Active: Yes
3. Women's Summer Dress
- Price: $39.99
- Stock Quantity: 80
- Launch Date: 2025-07-30
- Active: Yes
4. Cotton Hoodie Unisex
- Price: $54.99
- Stock Quantity: 40
- Launch Date: 2025-07-05
- Active: Yes
5. Slim Fit Chinos
- Price: $64.90
- Stock Quantity: 35
- Launch Date: 2025-07-22
- Active: Yes
6. Waterproof Rain Jacket
- Price: $99.50
- Stock Quantity: 22
- Launch Date: 2025-07-18
- Active: Yes
7. Graphic Tee Urban Style
- Price: $24.99
- Stock Quantity: 70
- Launch Date: 2025-08-03
- Active: Yes
If you need more details or want to take action on any of these products, please let me know.
Next, the following results are related to data updates.
user_input = "Update the price of the noise cancelling headphones to 229.90 and set the stock to 100 units."
Response: The product "Noise Cancelling Headphones" (ID: 7) has been successfully updated with a new price of 229.90 and a stock quantity of 100 units. If you need further assistance with this or any other product, please let me know!
user_input = "Change the category of the Python Programming book to 'Education' and mark it as inactive."
Response: The product "Python Programming for Beginners" has been successfully updated: its category is now set to "Education" and it is marked as inactive.
The next set of tests is related to data deletion.
user_input = "Please delete the product called classic white T-Shirt from the inventory."
Response: The product "Classic White T-Shirt" (ID: 2) has been successfully deleted from the inventory.
user_input = "Remove that home security camera we added recently."
Response: The product named 'Home Security Camera' has been successfully deleted from the inventory.
Finally, we will run some metrics tests. It’s important to note that, in order to reproduce the exact results shown below, you must first perform the update and delete steps, since they modify the table data. Otherwise, your results may differ, but you can still validate them through the SQL queries I’ll demonstrate next.
user_input = "What's the average price of products in the Electronics category?"
Response: The average price of products in the "Electronics" category is 112.88.
user_input = "What's the total inventory value across all products?"
Response: The total inventory value across all products is 65,626.0 dollars.
user_input = "Can you calculate the weighted average price of all products based on stock quantity?"
Response: The weighted average price of all products, based on their stock quantity, is 68.15.
Below are the corresponding SQL queries to validate each result, in the same order as above:
SELECT ROUND(AVG(price), 2) AS average_price
FROM products
WHERE category = 'Electronics';
SELECT SUM(price * stock_qty) AS total_inventory
FROM products;
SELECT ROUND(SUM(price * stock_qty) / SUM(stock_qty), 2) AS weighted_average_price
FROM products;
Edvaldo Melo 
Leave a comment