A strategy to standardize communication between AI functions and exterior instruments or knowledge sources. This standardization helps cut back the variety of integrations required (n*m to n+m):
- You should use a community-made MCP server whenever you want frequent performance. It saves time and avoids the necessity to reinvent the wheel each time.
- It’s also possible to publish your individual instruments and sources in order that others can use them.
In my earlier article, I created an Analytics Toolbox, a group of instruments that would doubtlessly automate each day routines. I constructed an MCP server and used options with present shoppers such because the MCP Inspector and Claude Desktop.
Now we wish to use these instruments instantly in our AI functions. To do this, construct your individual MCP consumer. Write a reasonably low stage code. This additionally provides you a transparent image of how instruments like Claude Code work together with the MCP underneath the hood.
Moreover, I wish to implement the present performance (July 2025) Lacking from Claude Desktop: Means to robotically verify and use LLM if it has a immediate template appropriate for the duty at hand. At present, it’s good to manually choose a template, which isn’t very helpful.
As a bonus, we additionally share a high-level implementation utilizing the Smolagents Framework. That is finest when working with solely MCP instruments and doesn’t require a lot customization.
MCP Protocol Overview
Here’s a temporary abstract of the MCP to be sure you are on the identical web page: MCP is a protocol developed by humanity to standardize the best way LLMS interacts with the skin world.
It follows a client-server structure and consists of three principal parts:
- host That is an software for customers.
- MCP Consumer Elements inside a number set up a one-to-one reference to the server and talk utilizing messages outlined within the MCP protocol.
- MCP Server Publish options akin to immediate templates, sources, instruments and extra.
As a result of we have already got it Implemented an MCP server Beforehand, this time we are going to give attention to constructing an MCP consumer. Beginning with a comparatively easy implementation, add the flexibility to dynamically choose immediate templates on the fly later.
You will discover the whole code github.
Constructing an MCP chatbot
Let’s begin with the primary setup: Load human API keys from the configuration file and alter Python asyncio An occasion loop that helps nested occasion loops.
# Load configuration and atmosphere
with open('../../config.json') as f:
config = json.load(f)
os.environ["ANTHROPIC_API_KEY"] = config['ANTHROPIC_API_KEY']
nest_asyncio.apply()Begin by constructing a program skeleton and acquire a transparent understanding of the high-level structure of your software.
async def principal():
"""Predominant entry level for the MCP ChatBot software."""
chatbot = MCP_ChatBot()
strive:
await chatbot.connect_to_servers()
await chatbot.chat_loop()
lastly:
await chatbot.cleanup()
if __name__ == "__main__":
asyncio.run(principal())Begin by creating an occasion from the start MCP_ChatBot class. The chatbot begins by discovering out there MCP options (it repeats all configured MCP servers, establishes a connection and requests an inventory of options).
As soon as the connection is ready up, the chatbot listens to the consumer queries, invokes instruments if essential, initializing an infinite loop the place this cycle continues till the method stops manually.
Lastly, carry out a cleanup step to shut all open connections.
Subsequent, let’s go into extra element about every stage.
Initializing a chatbot class
Let’s begin by creating and defining lessons __init__ technique. The principle fields of the chatbot class are:
exit_stackManages the lifecycle of a number of asynchronous threads (connections to MCP servers) to make sure that all connections are correctly closed whereas they’re working. This logic is carried out incleanupperform.anthropicA consumer of a synthetic API used to ship messages to LLM.available_toolsandavailable_promptsAn inventory of instruments and prompts printed by all linked MCP servers.periodsMapping instruments, prompts and sources to every MCP session. This enables the chatbot to route requests to the proper MCP server when LLM selects a particular software.
class MCP_ChatBot:
"""
MCP (Mannequin Context Protocol) ChatBot that connects to a number of MCP servers
and offers a conversational interface utilizing Anthropic's Claude.
Helps instruments, prompts, and sources from linked MCP servers.
"""
def __init__(self):
self.exit_stack = AsyncExitStack()
self.anthropic = Anthropic() # Consumer for Anthropic API
self.available_tools = [] # Instruments from all linked servers
self.available_prompts = [] # Prompts from all linked servers
self.periods = {} # Maps software/immediate/useful resource names to MCP periods
async def cleanup(self):
"""Clear up sources and shut all connections."""
await self.exit_stack.aclose()Hook up with the server
The primary activity of a chatbot is to provoke a reference to all configured MCP servers and uncover the options out there.
The record of MCP servers that the agent can connect with is server_config.json file. I arrange a connection on three MCP servers.
- Analyst_toolkit That is the implementation of the on a regular basis evaluation instruments defined within the earlier article.
- File System Makes the file out there to the agent.
- fetch LLMS can retrieve the content material of an internet web page and convert it from HTML to Markdown for simpler readability.
{
"mcpServers": {
"analyst_toolkit": {
"command": "uv",
"args": [
"--directory",
"/path/to/github/mcp-analyst-toolkit/src/mcp_server",
"run",
"server.py"
],
"env": {
"GITHUB_TOKEN": "your_github_token"
}
},
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/Users/marie/Desktop",
"/Users/marie/Documents/github"
]
},
"fetch": {
"command": "uvx",
"args": ["mcp-server-fetch"]
}
}
}First, learn and parse the configuration file earlier than connecting to every of the listed servers.
async def connect_to_servers(self):
"""Load server configuration and connect with all configured MCP servers."""
strive:
with open("server_config.json", "r") as file:
knowledge = json.load(file)
servers = knowledge.get("mcpServers", {})
for server_name, server_config in servers.gadgets():
await self.connect_to_server(server_name, server_config)
besides Exception as e:
print(f"Error loading server config: {e}")
traceback.print_exc()
elevateFor every server, carry out a number of steps to determine a connection.
- On the transport stage, we Begins the MCP server as an STDIO course of and retrieves the streams to ship and obtain messages.
- On the session stageCreate a
ClientSessionEmbody the stream after which name and carry out a handshake of the MCPinitializetechnique. - Registered each session and transport object with context supervisor
exit_stackMake sure that all connections are ultimately closed correctly. - The ultimate step is as follows: Register the server perform. This perform is wrapped in one other perform. I will clarify it quickly.
async def connect_to_server(self, server_name, server_config):
"""Hook up with a single MCP server and register its capabilities."""
strive:
server_params = StdioServerParameters(**server_config)
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params)
)
learn, write = stdio_transport
session = await self.exit_stack.enter_async_context(
ClientSession(learn, write)
)
await session.initialize()
await self._register_server_capabilities(session, server_name)
besides Exception as e:
print(f"Error connecting to {server_name}: {e}")
traceback.print_exc()The registration function contains repeating all instruments, prompts and sources retrieved from the session. Because of this, updates the interior variables periods (Mapping between sources and particular periods between MCP shoppers and servers),, available_prompts and available_tools.
async def _register_server_capabilities(self, session, server_name):
"""Register instruments, prompts and sources from a single server."""
capabilities = [
("tools", session.list_tools, self._register_tools),
("prompts", session.list_prompts, self._register_prompts),
("resources", session.list_resources, self._register_resources)
]
for capability_name, list_method, register_method in capabilities:
strive:
response = await list_method()
await register_method(response, session)
besides Exception as e:
print(f"Server {server_name} does not help {capability_name}: {e}")
async def _register_tools(self, response, session):
"""Register instruments from server response."""
for software in response.instruments:
self.periods[tool.name] = session
self.available_tools.append({
"title": software.title,
"description": software.description,
"input_schema": software.inputSchema
})
async def _register_prompts(self, response, session):
"""Register prompts from server response."""
if response and response.prompts:
for immediate in response.prompts:
self.periods[prompt.name] = session
self.available_prompts.append({
"title": immediate.title,
"description": immediate.description,
"arguments": immediate.arguments
})
async def _register_resources(self, response, session):
"""Register sources from server response."""
if response and response.sources:
for useful resource in response.sources:
resource_uri = str(useful resource.uri)
self.periods[resource_uri] = sessionBy the top of this stage, we MCP_ChatBot An object has every part it’s good to begin interacting with the consumer.
- Connections to all configured MCP servers are established.
- All prompts, sources, and instruments are registered, together with the directions wanted to grasp how LLM makes use of these options.
- The mapping between these sources and every session is saved, so you recognize precisely the place to ship every request.
Chat loop
So it is time to begin chatting with customers chat_loop perform.
First, share all out there instructions with the consumer.
- An inventory of sources, instruments, and prompts
- Operating software calls
- Viewing sources
- Use the immediate template
- Cease chatting (It is necessary that there’s a clear strategy to finish an infinite loop).
It then performs the suitable motion based mostly on the consumer’s enter. Do one of many above instructions or whether or not to make a request to LLM.
async def chat_loop(self):
"""Predominant interactive chat loop with command processing."""
print("nMCP Chatbot Began!")
print("Instructions:")
print(" give up - Exit the chatbot")
print(" @durations - Present out there changelog durations")
print(" @<interval> - View changelog for particular interval")
print(" /instruments - Checklist out there instruments")
print(" /software <title> <arg1=value1> - Execute a software with arguments")
print(" /prompts - Checklist out there prompts")
print(" /immediate <title> <arg1=value1> - Execute a immediate with arguments")
whereas True:
strive:
question = enter("nQuery: ").strip()
if not question:
proceed
if question.decrease() == 'give up':
break
# Deal with useful resource requests (@command)
if question.startswith('@'):
interval = question[1:]
resource_uri = "changelog://durations" if interval == "durations" else f"changelog://{interval}"
await self.get_resource(resource_uri)
proceed
# Deal with slash instructions
if question.startswith('/'):
components = self._parse_command_arguments(question)
if not components:
proceed
command = components[0].decrease()
if command == '/instruments':
await self.list_tools()
elif command == '/software':
if len(components) < 2:
print("Utilization: /software <title> <arg1=value1> <arg2=value2>")
proceed
tool_name = components[1]
args = self._parse_prompt_arguments(components[2:])
await self.execute_tool(tool_name, args)
elif command == '/prompts':
await self.list_prompts()
elif command == '/immediate':
if len(components) < 2:
print("Utilization: /immediate <title> <arg1=value1> <arg2=value2>")
proceed
prompt_name = components[1]
args = self._parse_prompt_arguments(components[2:])
await self.execute_prompt(prompt_name, args)
else:
print(f"Unknown command: {command}")
proceed
# Course of common queries
await self.process_query(question)
besides Exception as e:
print(f"nError in chat loop: {e}")
traceback.print_exc()There are a lot of helper capabilities to parse arguments and return an inventory of beforehand registered out there instruments and prompts. It is fairly easy, so I will not go into an excessive amount of element right here. You possibly can verify Complete code In case you’re .
As a substitute, let’s dig deeper into how the interplay between an MCP consumer and a server works in numerous situations.
When utilizing sources, we use them self.periods Map to search out the suitable session (utilizing the fallback choice if essential), and use that session to learn the useful resource.
async def get_resource(self, resource_uri):
"""Retrieve and show content material from an MCP useful resource."""
session = self.periods.get(resource_uri)
# Fallback: discover any session that handles this useful resource kind
if not session and resource_uri.startswith("changelog://"):
session = subsequent(
(sess for uri, sess in self.periods.gadgets()
if uri.startswith("changelog://")),
None
)
if not session:
print(f"Useful resource '{resource_uri}' not discovered.")
return
strive:
end result = await session.read_resource(uri=resource_uri)
if end result and end result.contents:
print(f"nResource: {resource_uri}")
print("Content material:")
print(end result.contents[0].textual content)
else:
print("No content material out there.")
besides Exception as e:
print(f"Error studying useful resource: {e}")
traceback.print_exc()To run the software, comply with an analogous course of. Discover the session after which use it to invoke the software and go its title and arguments.
async def execute_tool(self, tool_name, args):
"""Execute an MCP software instantly with given arguments."""
session = self.periods.get(tool_name)
if not session:
print(f"Software '{tool_name}' not discovered.")
return
strive:
end result = await session.call_tool(tool_name, arguments=args)
print(f"nTool '{tool_name}' end result:")
print(end result.content material)
besides Exception as e:
print(f"Error executing software: {e}")
traceback.print_exc()There are not any surprises right here. The identical strategy works for working the immediate.
async def execute_prompt(self, prompt_name, args):
"""Execute an MCP immediate with given arguments and course of the end result."""
session = self.periods.get(prompt_name)
if not session:
print(f"Immediate '{prompt_name}' not discovered.")
return
strive:
end result = await session.get_prompt(prompt_name, arguments=args)
if end result and end result.messages:
prompt_content = end result.messages[0].content material
textual content = self._extract_prompt_text(prompt_content)
print(f"nExecuting immediate '{prompt_name}'...")
await self.process_query(textual content)
besides Exception as e:
print(f"Error executing immediate: {e}")
traceback.print_exc()The one main use case that we have not lined but is to deal with common freeform enter from customers (not any of the precise instructions).
On this case, you first ship the primary request to LLM, then parse the output to outline whether or not there’s a software name. If there are software calls, run them. In any other case, it ends the infinite loop and returns the reply to the consumer.
async def process_query(self, question):
"""Course of a consumer question by Anthropic's Claude, dealing with software calls iteratively."""
messages = [{'role': 'user', 'content': query}]
whereas True:
response = self.anthropic.messages.create(
max_tokens=2024,
mannequin='claude-3-7-sonnet-20250219',
instruments=self.available_tools,
messages=messages
)
assistant_content = []
has_tool_use = False
for content material in response.content material:
if content material.kind == 'textual content':
print(content material.textual content)
assistant_content.append(content material)
elif content material.kind == 'tool_use':
has_tool_use = True
assistant_content.append(content material)
messages.append({'function': 'assistant', 'content material': assistant_content})
# Execute the software name
session = self.periods.get(content material.title)
if not session:
print(f"Software '{content material.title}' not discovered.")
break
end result = await session.call_tool(content material.title, arguments=content material.enter)
messages.append({
"function": "consumer",
"content material": [{
"type": "tool_result",
"tool_use_id": content.id,
"content": result.content
}]
})
if not has_tool_use:
breakSo we have absolutely lined how the MCP chatbot truly works underneath the hood. Now it is time to work and take a look at it. You possibly can run it from the command line interface utilizing the next command:
python mcp_client_example_base.pyWhenever you run the chatbot, you’ll first obtain the next introduction message:
MCP Chatbot Began!
Instructions:
give up - Exit the chatbot
@durations - Present out there changelog durations
@<interval> - View changelog for particular interval
/instruments - Checklist out there instruments
/software <title> <arg1=value1> - Execute a software with arguments
/prompts - Checklist out there prompts
/immediate <title> <arg1=value1> - Execute a immediate with argumentsFrom there, you’ll be able to strive totally different instructions, for instance.
- Name the software to record the databases out there within the DB
- Checklist all out there prompts
- Utilizing the immediate template, name it like this
/immediate sql_query_prompt query=”What number of clients did now we have in Might 2024?”.
Lastly, you’ll be able to enter and finish the chat give up.
Question: /software list_databases
[07/02/25 18:27:28] INFO Processing request of kind CallToolRequest server.py:619
Software 'list_databases' end result:
[TextContent(type='text', text='INFORMATION_SCHEMAndatasetsndefaultnecommercenecommerce_dbninformation_schemansystemn', annotations=None, meta=None)]
Question: /prompts
Obtainable prompts:
- sql_query_prompt: Create a SQL question immediate
Arguments:
- query
Question: /immediate sql_query_prompt query="What number of clients did now we have in Might 2024?"
[07/02/25 18:28:21] INFO Processing request of kind GetPromptRequest server.py:619
Executing immediate 'sql_query_prompt'...
I will create a SQL question to search out the variety of clients in Might 2024.
[07/02/25 18:28:25] INFO Processing request of kind CallToolRequest server.py:619
Based mostly on the question outcomes, this is the ultimate SQL question:
```sql
choose uniqExact(user_id) as customer_count
from ecommerce.periods
the place toStartOfMonth(action_date) = '2024-05-01'
format TabSeparatedWithNames
```
Question: /software execute_sql_query question="choose uniqExact(user_id) as customer_count from ecommerce.periods the place toStartOfMonth(action_date) = '2024-05-01' format TabSeparatedWithNames"
I will assist you execute this SQL question to get the distinctive buyer depend for Might 2024. Let me run this for you.
[07/02/25 18:30:09] INFO Processing request of kind CallToolRequest server.py:619
The question has been executed efficiently. The outcomes present that there have been 246,852 distinctive clients (distinctive user_ids) in Might 2024 based mostly on the ecommerce.periods desk.
Question: give upIt appears to be like fairly cool! Our fundamental model works properly! Now, it is time to take it a step additional by instructing clients to recommend related prompts on the spot based mostly on enter from clients, and make your chatbot smarter.
Fast options
The truth is, suggesting one of the best immediate template to your activity might be extraordinarily useful. At present, our chatbot customers ought to already know in regards to the out there prompts, or no less than have an interest sufficient to discover on their very own to profit from what now we have constructed. Including a fast suggestion function lets you make this discovery to your customers and make your chatbot extra handy and user-friendly.
Let’s brainstorm the best way to add this function. Strategy this function within the following method:
Consider immediate relevance utilizing LLM. Repeat all out there immediate templates to evaluate for each that the immediate is appropriate for the consumer’s question.
Recommend a immediate that matches the consumer. In case you discover a associated immediate template, share it with the consumer and ask if you wish to run it.
Merge the immediate template with consumer enter. If the consumer accepts it, mix the chosen immediate with the unique question. The immediate template has a placeholder, so you could want LLM to enter it. After you have merged the immediate template together with your question, you’ll create an replace message able to ship to LLM.
Add this logic to process_query perform. Because of the modular design, it is very straightforward so as to add this extension with out complicated the remainder of your code.
Let’s begin by implementing the perform and discovering probably the most related immediate template. Consider every immediate utilizing LLM and assign a associated rating from 0 to five. It then filters out prompts with scores of two or much less, returning solely probably the most related (the one with the very best associated rating of the remaining outcomes).
async def _find_matching_prompt(self, question):
"""Discover a matching immediate for the given question utilizing LLM analysis."""
if not self.available_prompts:
return None
# Use LLM to guage immediate relevance
prompt_scores = []
for immediate in self.available_prompts:
# Create analysis immediate for the LLM
evaluation_prompt = f"""
You might be an knowledgeable at evaluating whether or not a immediate template is related for a consumer question.
Consumer Question: "{question}"
Immediate Template:
- Identify: {immediate['name']}
- Description: {immediate['description']}
Fee the relevance of this immediate template for the consumer question on a scale of 0-5:
- 0: Fully irrelevant
- 1: Barely related
- 2: Considerably related
- 3: Reasonably related
- 4: Extremely related
- 5: Excellent match
Contemplate:
- Does the immediate template handle the consumer's intent?
- Would utilizing this immediate template present a greater response than a generic question?
- Are the matters and context aligned?
Reply with solely a single quantity (0-5) and no different textual content.
"""
strive:
response = self.anthropic.messages.create(
max_tokens=10,
mannequin='claude-3-7-sonnet-20250219',
messages=[{'role': 'user', 'content': evaluation_prompt}]
)
# Extract the rating from the response
score_text = response.content material[0].textual content.strip()
rating = int(score_text)
if rating >= 3: # Solely take into account prompts with rating >= 3
prompt_scores.append((immediate, rating))
besides Exception as e:
print(f"Error evaluating immediate {immediate['name']}: {e}")
proceed
# Return the immediate with the very best rating
if prompt_scores:
best_prompt, best_score = max(prompt_scores, key=lambda x: x[1])
return best_prompt
return NoneThe next capabilities that have to be carried out are the mix of the chosen immediate template and consumer enter: Depend on LLM to intelligently mix them to fill in all placeholders as wanted.
async def _combine_prompt_with_query(self, prompt_name, user_query):
"""Use LLM to mix immediate template with consumer question."""
# First, get the immediate template content material
session = self.periods.get(prompt_name)
if not session:
print(f"Immediate '{prompt_name}' not discovered.")
return None
strive:
# Discover the immediate definition to get its arguments
prompt_def = None
for immediate in self.available_prompts:
if immediate['name'] == prompt_name:
prompt_def = immediate
break
# Put together arguments for the immediate template
args = {}
if prompt_def and prompt_def.get('arguments'):
for arg in prompt_def['arguments']:
arg_name = arg.title if hasattr(arg, 'title') else arg.get('title', '')
if arg_name:
# Use placeholder format for arguments
args[arg_name] = '<' + str(arg_name) + '>'
# Get the immediate template with arguments
end result = await session.get_prompt(prompt_name, arguments=args)
if not end result or not end result.messages:
print(f"Couldn't retrieve immediate template for '{prompt_name}'")
return None
prompt_content = end result.messages[0].content material
prompt_text = self._extract_prompt_text(prompt_content)
# Create mixture immediate for the LLM
combination_prompt = f"""
You might be an knowledgeable at combining immediate templates with consumer queries to create optimized prompts.
Authentic Consumer Question: "{user_query}"
Immediate Template:
{prompt_text}
Your activity:
1. Analyze the consumer's question and the immediate template
2. Mix them intelligently to create a single, coherent immediate
3. Make sure the consumer's particular query/request is addressed inside the context of the template
4. Keep the construction and intent of the template whereas incorporating the consumer's question
Reply with solely the mixed immediate textual content, no explanations or extra textual content.
"""
response = self.anthropic.messages.create(
max_tokens=2048,
mannequin='claude-3-7-sonnet-20250219',
messages=[{'role': 'user', 'content': combination_prompt}]
)
return response.content material[0].textual content.strip()
besides Exception as e:
print(f"Error combining immediate with question: {e}")
return NoneThen merely replace process_query To substantiate the logic matching immediate, ask the consumer to substantiate and resolve which messages to ship to LLM.
async def process_query(self, question):
"""Course of a consumer question by Anthropic's Claude, dealing with software calls iteratively."""
# Test if there is a matching immediate first
matching_prompt = await self._find_matching_prompt(question)
if matching_prompt:
print(f"Discovered matching immediate: {matching_prompt['name']}")
print(f"Description: {matching_prompt['description']}")
# Ask consumer in the event that they wish to use the immediate template
use_prompt = enter("Would you want to make use of this immediate template? (y/n): ").strip().decrease()
if use_prompt == 'y' or use_prompt == 'sure':
print("Combining immediate template together with your question...")
# Use LLM to mix immediate template with consumer question
combined_prompt = await self._combine_prompt_with_query(matching_prompt['name'], question)
if combined_prompt:
print(f"Mixed immediate created. Processing...")
# Course of the mixed immediate as an alternative of the unique question
messages = [{'role': 'user', 'content': combined_prompt}]
else:
print("Failed to mix immediate template. Utilizing unique question.")
messages = [{'role': 'user', 'content': query}]
else:
# Use unique question if consumer does not wish to use the immediate
messages = [{'role': 'user', 'content': query}]
else:
# Course of the unique question if no matching immediate discovered
messages = [{'role': 'user', 'content': query}]
# print(messages)
# Course of the ultimate question (both unique or mixed)
whereas True:
response = self.anthropic.messages.create(
max_tokens=2024,
mannequin='claude-3-7-sonnet-20250219',
instruments=self.available_tools,
messages=messages
)
assistant_content = []
has_tool_use = False
for content material in response.content material:
if content material.kind == 'textual content':
print(content material.textual content)
assistant_content.append(content material)
elif content material.kind == 'tool_use':
has_tool_use = True
assistant_content.append(content material)
messages.append({'function': 'assistant', 'content material': assistant_content})
# Log software name data
print(f"n[TOOL CALL] Software: {content material.title}")
print(f"[TOOL CALL] Arguments: {json.dumps(content material.enter, indent=2)}")
# Execute the software name
session = self.periods.get(content material.title)
if not session:
print(f"Software '{content material.title}' not discovered.")
break
end result = await session.call_tool(content material.title, arguments=content material.enter)
# Log software end result
print(f"[TOOL RESULT] Software: {content material.title}")
print(f"[TOOL RESULT] Content material: {end result.content material}")
messages.append({
"function": "consumer",
"content material": [{
"type": "tool_result",
"tool_use_id": content.id,
"content": result.content
}]
})
if not has_tool_use:
breakSubsequent, let’s take a look at the up to date model together with your knowledge query. Excitingly, the chatbot was capable of finding the proper immediate and use it to search out the proper reply.
Question: What number of clients did now we have in Might 2024?
Discovered matching immediate: sql_query_prompt
Description: Create a SQL question immediate
Would you want to make use of this immediate template? (y/n): y
Combining immediate template together with your question...
[07/05/25 14:38:58] INFO Processing request of kind GetPromptRequest server.py:619
Mixed immediate created. Processing...
I will write a question to depend distinctive clients who had periods in Might 2024. Since this can be a enterprise metric, I will exclude fraudulent periods.
[TOOL CALL] Software: execute_sql_query
[TOOL CALL] Arguments: {
"question": "/* Rely distinct customers with non-fraudulent periods in Might 2024n Utilizing uniqExact for exact consumer countn Filtering for Might 2024 utilizing toStartOfMonth and including date vary */nSELECT n uniqExactIf(s.user_id, s.is_fraud = 0) AS active_customers_countnFROM ecommerce.periods snWHERE toStartOfMonth(action_date) = toDate('2024-05-01')nFORMAT TabSeparatedWithNames"
}
[07/05/25 14:39:17] INFO Processing request of kind CallToolRequest server.py:619
[TOOL RESULT] Software: execute_sql_query
[TOOL RESULT] Content material: [TextContent(type='text', text='active_customers_countn245287n', annotations=None, meta=None)]
The question reveals we had 245,287 distinctive clients with reputable (non-fraudulent) periods in Might 2024. Here is a breakdown of why I wrote the question this fashion:
1. Used uniqExactIf() to get exact depend of distinctive customers whereas excluding fraudulent periods in a single step
2. Used toStartOfMonth() to make sure we seize all days in Might 2024
3. Specified the date format correctly with toDate('2024-05-01')
4. Used TabSeparatedWithNames format as required
5. Offered a significant column alias
Would you prefer to see any variations of this evaluation, akin to together with fraudulent periods or breaking down the numbers by nation?Testing detrimental examples is at all times a good suggestion. On this case, the chatbot works as anticipated and doesn’t recommend a SQL-related immediate if given an unrelated query.
Question: How are you?
I ought to be aware that I am an AI assistant centered on serving to you're employed with the out there instruments, which embody executing SQL queries, getting database/desk data, and accessing GitHub PR knowledge. I haven't got a software particularly for responding to private questions.
I might help you:
- Question a ClickHouse database
- Checklist databases and describe tables
- Get details about GitHub Pull Requests
What would you prefer to learn about these areas?With the chatbot working, we’re able to put issues collectively.
Bonus: Fast and simple MCP consumer with Smolagents
I’ve regarded into low-level code that permits for constructing extremely custom-made MCP shoppers, however many use circumstances require solely fundamental performance. So we determined to share a fast and easy implementation for the situation when solely the instruments are wanted. I exploit one in all my favourite agent frameworks – Smolagents in Huggingface (I’ve defined this framework intimately My earlier publish).
# wanted imports
from smolagents import CodeAgent, DuckDuckGoSearchTool, LiteLLMModel, VisitWebpageTool, ToolCallingAgent, ToolCollection
from mcp import StdioServerParameters
import json
import os
# setting OpenAI APIKey
with open('../../config.json') as f:
config = json.masses(f.learn())
os.environ["OPENAI_API_KEY"] = config['OPENAI_API_KEY']
# defining the LLM
mannequin = LiteLLMModel(
model_id="openai/gpt-4o-mini",
max_tokens=2048
)
# configuration for the MCP server
server_parameters = StdioServerParameters(
command="uv",
args=[
"--directory",
"/path/to/github/mcp-analyst-toolkit/src/mcp_server",
"run",
"server.py"
],
env={"GITHUB_TOKEN": "github_<your_token>"},
)
# immediate
CLICKHOUSE_PROMPT_TEMPLATE = """
You're a senior knowledge analyst with greater than 10 years of expertise writing complicated SQL queries, particularly optimized for ClickHouse to reply consumer questions.
## Database Schema
You might be working with an e-commerce analytics database containing the next tables:
### Desk: ecommerce.customers
**Description:** Buyer data for the web store
**Main Key:** user_id
**Fields:**
- user_id (Int64) - Distinctive buyer identifier (e.g., 1000004, 3000004)
- nation (String) - Buyer's nation of residence (e.g., "Netherlands", "United Kingdom")
- is_active (Int8) - Buyer standing: 1 = lively, 0 = inactive
- age (Int32) - Buyer age in full years (e.g., 31, 72)
### Desk: ecommerce.periods
**Description:** Consumer session knowledge and transaction data
**Main Key:** session_id
**International Key:** user_id (references ecommerce.customers.user_id)
**Fields:**
- user_id (Int64) - Buyer identifier linking to customers desk (e.g., 1000004, 3000004)
- session_id (Int64) - Distinctive session identifier (e.g., 106, 1023)
- action_date (Date) - Session begin date (e.g., "2021-01-03", "2024-12-02")
- session_duration (Int32) - Session period in seconds (e.g., 125, 49)
- os (String) - Working system used (e.g., "Home windows", "Android", "iOS", "MacOS")
- browser (String) - Browser used (e.g., "Chrome", "Safari", "Firefox", "Edge")
- is_fraud (Int8) - Fraud indicator: 1 = fraudulent session, 0 = reputable
- income (Float64) - Buy quantity in USD (0.0 for non-purchase periods, >0 for purchases)
## ClickHouse-Particular Pointers
1. **Use ClickHouse-optimized capabilities:**
- uniqExact() for exact distinctive counts
- uniqExactIf() for conditional distinctive counts
- quantile() capabilities for percentiles
- Date capabilities: toStartOfMonth(), toStartOfYear(), right this moment()
2. **Question formatting necessities:**
- At all times finish queries with "format TabSeparatedWithNames"
- Use significant column aliases
- Use correct JOIN syntax when combining tables
- Wrap date literals in quotes (e.g., '2024-01-01')
3. **Efficiency concerns:**
- Use applicable WHERE clauses to filter knowledge
- Think about using HAVING for post-aggregation filtering
- Use LIMIT when discovering high/backside outcomes
4. **Information interpretation:**
- income > 0 signifies a purchase order session
- income = 0 signifies a searching session with out buy
- is_fraud = 1 periods ought to sometimes be excluded from enterprise metrics except particularly analyzing fraud
## Response Format
Present solely the SQL question as your reply. Embody temporary reasoning in feedback if the question logic is complicated.
## Examples
**Query:** What number of clients made buy in December 2024?
**Reply:** choose uniqExact(user_id) as clients from ecommerce.periods the place toStartOfMonth(action_date) = '2024-12-01' and income > 0 format TabSeparatedWithNames
**Query:** What was the fraud price in 2023, expressed as a proportion?
**Reply:** choose 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.periods the place toStartOfYear(action_date) = '2023-01-01' format TabSeparatedWithNames
**Query:** What was the share of customers utilizing Home windows yesterday?
**Reply:** choose 100 * uniqExactIf(user_id, os = 'Home windows') / uniqExact(user_id) as windows_share from ecommerce.periods the place action_date = right this moment() - 1 format TabSeparatedWithNames
**Query:** What was the income from Dutch customers aged 55 and older in December 2024?
**Reply:** choose sum(s.income) as total_revenue from ecommerce.periods as s internal be a part of ecommerce.customers as u on s.user_id = u.user_id the place u.nation = 'Netherlands' and u.age >= 55 and toStartOfMonth(s.action_date) = '2024-12-01' format TabSeparatedWithNames
**Query:** What are the median and interquartile vary (IQR) of buy income for every nation?
**Reply:** choose nation, median(income) as median_revenue, quantile(0.25)(income) as q25_revenue, quantile(0.75)(income) as q75_revenue from ecommerce.periods as s internal be a part of ecommerce.customers as u on u.user_id = s.user_id the place income > 0 group by nation format TabSeparatedWithNames
**Query:** What's the common variety of days between the primary session and the primary buy for customers who made no less than one buy?
**Reply:** choose avg(first_purchase - first_action_date) as avg_days_to_purchase from (choose user_id, min(action_date) as first_action_date, minIf(action_date, income > 0) as first_purchase, max(income) as max_revenue from ecommerce.periods group by user_id) the place max_revenue > 0 format TabSeparatedWithNames
**Query:** What's the variety of periods in December 2024, damaged down by working methods, together with the totals?
**Reply:** choose os, uniqExact(session_id) as session_count from ecommerce.periods the place toStartOfMonth(action_date) = '2024-12-01' group by os with totals format TabSeparatedWithNames
**Query:** Do now we have clients who used a number of browsers throughout 2024? In that case, please calculate the variety of clients for every mixture of browsers.
**Reply:** choose browsers, depend(*) as customer_count from (choose user_id, arrayStringConcat(arraySort(groupArray(distinct browser)), ', ') as browsers from ecommerce.periods the place toStartOfYear(action_date) = '2024-01-01' group by user_id) group by browsers order by customer_count desc format TabSeparatedWithNames
**Query:** Which browser has the very best share of fraud customers?
**Reply:** choose browser, 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.periods group by browser order by fraud_rate desc restrict 1 format TabSeparatedWithNames
**Query:** Which nation had the very best variety of first-time customers in 2024?
**Reply:** choose nation, depend(distinct user_id) as new_users from (choose user_id, min(action_date) as first_date from ecommerce.periods group by user_id having toStartOfYear(first_date) = '2024-01-01') as t internal be a part of ecommerce.customers as u on t.user_id = u.user_id group by nation order by new_users desc restrict 1 format TabSeparatedWithNames
---
**Your Activity:** Utilizing all of the offered data above, write a ClickHouse SQL question to reply the next buyer query:
{query}
"""
with ToolCollection.from_mcp(server_parameters, trust_remote_code=True) as tool_collection:
agent = ToolCallingAgent(instruments=[*tool_collection.tools], mannequin=mannequin)
immediate = CLICKHOUSE_PROMPT_TEMPLATE.format(
query = 'What number of clients did now we have in Might 2024?'
)
response = agent.run(immediate)Because of this, I acquired the proper reply.
In case you do not want a lot customization or integration with prompts or sources, this implementation is certainly a strategy to proceed.
abstract
On this article, now we have created a chatbot that integrates with an MCP server and takes benefit of all the advantages of standardization to seamlessly entry instruments, prompts and sources.
I began with a fundamental implementation that permits me to record and entry MCP performance. We then enhanced the chatbot with a sensible function that implies user-related immediate templates based mostly on enter. This makes our merchandise extra intuitive and user-friendly, particularly for customers who’re new to the whole library of accessible prompts.
To implement chatbots, you need to use comparatively low-level code to higher perceive how the MCP protocol works underneath the hood, and what occurs when utilizing AI instruments like Claude Desktop and Cursor.
As a bonus, we additionally mentioned the implementation of Smolagents. This lets you shortly deploy MCP shoppers built-in with instruments.
Thanks for studying. I hope this text was insightful. Remember Einstein’s recommendation. “The necessary factor is to not cease asking questions. Curiosity has a purpose to exist.” Might your curiosity information you to the subsequent nice perception.
reference
This text is impressed by “MCP: Build a rich context AI app for humanity” A brief course from deeplearning.ai.
