Constructing a Сustom MCP Chatbot

a technique to standardise communication between AI applications and external tools or data sources. This standardisation helps to scale back the variety of integrations needed (): 

• You should use community-built MCP servers whenever you need common functionality, saving time and avoiding the necessity to reinvent the wheel each time.
• You can even expose your individual tools and resources, making them available for others to make use of.

In my previous article, we built the analytics toolbox (a set of tools that may automate your day-to-day routine). We built an MCP server and used its capabilities with existing clients like MCP Inspector or Claude Desktop. 

Now, we wish to make use of those tools directly in our AI applications. To try this, let’s construct our own MCP client. We’ll write fairly low-level code, which will even provide you with a clearer picture of how tools like Claude Code interact with MCP under the hood.

Moreover, I would really like to implement the feature that’s currently () missing from Claude Desktop: the power for the LLM to mechanically check whether it has an acceptable prompt template for the duty at hand and use it. Immediately, you might have to choose the template manually, which isn’t very convenient. 

As a bonus, I will even share a high-level implementation using the smolagents framework, which is right for scenarios whenever you work only with MCP tools and don’t need much customisation.

MCP protocol overview

Here’s a fast recap of the MCP to make sure we’re on the identical page. MCP is a protocol developed by Anthropic to standardise the way in which LLMs interact with the skin world. 

It follows a client-server architecture and consists of three essential components: 

• Host is the user-facing application. 
• MCP client is a component throughout the host that establishes a one-to-one reference to the server and communicates using messages defined by the MCP protocol.
• MCP server exposes capabilities reminiscent of prompt templates, resources and tools. 

Since we’ve already implemented the MCP server before, this time we are going to concentrate on constructing the MCP client. We’ll start with a comparatively easy implementation and later add the power to dynamically select prompt templates on the fly.

Constructing the MCP chatbot

Let’s begin with the initial setup: we’ll load the Anthropic API key from a config file and adjust Python’s asyncio event loop to support nested event loops.

# Load configuration and environment
with open('../../config.json') as f:
    config = json.load(f)
os.environ["ANTHROPIC_API_KEY"] = config['ANTHROPIC_API_KEY']

nest_asyncio.apply()

Let’s start by constructing a skeleton of our program to get a transparent picture of the appliance’s high-level architecture.

async def essential():
    """Major entry point for the MCP ChatBot application."""
    chatbot = MCP_ChatBot()
    try:
        await chatbot.connect_to_servers()
        await chatbot.chat_loop()
    finally:
        await chatbot.cleanup()

if __name__ == "__main__":
    asyncio.run(essential())

We start by creating an instance of the MCP_ChatBot class. The chatbot starts by discovering available MCP capabilities (iterating through all configured MCP servers, establishing connections and requesting their lists of capabilities). 

Once connections are arrange, we are going to initialise an infinite loop where the chatbot listens to the user queries, calls tools when needed and continues this cycle until the method is stopped manually. 

Finally, we are going to perform a cleanup step to shut all open connections.

Let’s now walk through each stage in additional detail.

Initialising the ChatBot class

Let’s start by creating the category and defining the __init__ method. The essential fields of the ChatBot class are: 

• exit_stack manages the lifecycle of multiple async threads (connections to MCP servers), ensuring that every one connections can be closed appropriately, even when we face an error during execution. This logic is implemented within the cleanup function.
• anthropic is a client for Anthropic API used to send messages to LLM.
• available_tools and available_prompts are the lists of tools and prompts exposed by all MCP servers we’re connected to. 
• sessions is a mapping of tools, prompts and resources to their respective MCP sessions. This enables the chatbot to route requests to the proper MCP server when the LLM selects a selected tool.

class MCP_ChatBot:
  """
  MCP (Model Context Protocol) ChatBot that connects to multiple MCP servers
  and provides a conversational interface using Anthropic's Claude.
    
  Supports tools, prompts, and resources from connected MCP servers.
  """
    
  def __init__(self):
    self.exit_stack = AsyncExitStack() 
    self.anthropic = Anthropic() # Client for Anthropic API
    self.available_tools = [] # Tools from all connected servers
    self.available_prompts = [] # Prompts from all connected servers  
    self.sessions = {} # Maps tool/prompt/resource names to MCP sessions

  async def cleanup(self):
    """Clean up resources and shut all connections."""
    await self.exit_stack.aclose()

Connecting to servers

The primary task for our chatbot is to initiate connections with all configured MCP servers and discover what capabilities we are able to use. 

The list of MCP servers that our agent can hook up with is defined within the server_config.json file. I’ve arrange connections with three MCP servers:

• analyst_toolkit is my implementation of the on a regular basis analytical tools we discussed within the previous article, 
• Filesystem allows the agent to work with files,
• Fetch helps LLMs retrieve the content of webpages and convert it from HTML to markdown for higher 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, we are going to read the config file, parse it after which hook up with each listed server.

async def connect_to_servers(self):
  """Load server configuration and hook up with all configured MCP servers."""
  try:
    with open("server_config.json", "r") as file:
      data = json.load(file)
    
    servers = data.get("mcpServers", {})
    for server_name, server_config in servers.items():
      await self.connect_to_server(server_name, server_config)
  except Exception as e:
    print(f"Error loading server config: {e}")
    traceback.print_exc()
    raise

For every server, we perform several steps to ascertain the connection:

• On the transport level, we launch the MCP server as a stdio process and get streams for sending and receiving messages. 
• On the session level, we create a ClientSession incorporating the streams, after which we perform the MCP handshake by calling initialize method.
• We registered each the session and transport objects within the context manager exit_stack to make sure that all connections can be closed properly ultimately. 
• The last step is to register server capabilities. We wrapped this functionality right into a separate function, and we are going to discuss it shortly.

async def connect_to_server(self, server_name, server_config):
    """Connect with a single MCP server and register its capabilities."""
    try:
      server_params = StdioServerParameters(**server_config)
      stdio_transport = await self.exit_stack.enter_async_context(
          stdio_client(server_params)
      )
      read, write = stdio_transport
      session = await self.exit_stack.enter_async_context(
          ClientSession(read, write)
      )
      await session.initialize()
      await self._register_server_capabilities(session, server_name)
            
    except Exception as e:
      print(f"Error connecting to {server_name}: {e}")
      traceback.print_exc()

Registering capabilities involves iterating over all of the tools, prompts and resources retrieved from the session. In consequence, we update the inner variables sessions (), available_prompts and available_tools.

async def _register_server_capabilities(self, session, server_name):
  """Register tools, prompts and resources 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:
    try:
      response = await list_method()
      await register_method(response, session)
    except Exception as e:
      print(f"Server {server_name} doesn't support {capability_name}: {e}")

async def _register_tools(self, response, session):
  """Register tools from server response."""
  for tool in response.tools:
    self.sessions[tool.name] = session
    self.available_tools.append({
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    })

async def _register_prompts(self, response, session):
  """Register prompts from server response."""
  if response and response.prompts:
    for prompt in response.prompts:
        self.sessions[prompt.name] = session
        self.available_prompts.append({
            "name": prompt.name,
            "description": prompt.description,
            "arguments": prompt.arguments
        })

async def _register_resources(self, response, session):
  """Register resources from server response."""
  if response and response.resources:
    for resource in response.resources:
        resource_uri = str(resource.uri)
        self.sessions[resource_uri] = session

By the tip of this stage, our MCP_ChatBot object has all the things it needs to start out interacting with users:

• connections to all configured MCP servers are established,
• all prompts, resources and tools are registered, including descriptions needed for LLM to grasp the best way to use these capabilities,
• mappings between these resources and their respective sessions are stored, so we all know exactly where to send each request.

Chat loop

So, it’s time to start out our chat with users by creating the chat_loop function. 

We’ll first share all of the available commands with the user: 

• listing resources, tools and prompts 
• executing a tool call 
• viewing a resource 
• using a prompt template
• quitting the chat ().

After that, we are going to enter an infinite loop where, based on user input, we are going to execute the suitable motion: whether it’s one among the commands above or making a request to the LLM.

async def chat_loop(self):
  """Major interactive chat loop with command processing."""
  print("nMCP Chatbot Began!")
  print("Commands:")
  print("  quit                           - Exit the chatbot")
  print("  @periods                       - Show available changelog periods") 
  print("  @                      - View changelog for specific period")
  print("  /tools                         - List available tools")
  print("  /tool       - Execute a tool with arguments")
  print("  /prompts                       - List available prompts")
  print("  /prompt     - Execute a prompt with arguments")
  
  while True:
    try:
      query = input("nQuery: ").strip()
      if not query:
          proceed

      if query.lower() == 'quit':
          break
      
      # Handle resource requests (@command)
      if query.startswith('@'):
        period = query[1:]
        resource_uri = "changelog://periods" if period == "periods" else f"changelog://{period}"
        await self.get_resource(resource_uri)
        proceed
      
      # Handle slash commands
      if query.startswith('/'):
        parts = self._parse_command_arguments(query)
        if not parts:
          proceed
            
        command = parts[0].lower()
        
        if command == '/tools':
          await self.list_tools()
        elif command == '/tool':
          if len(parts) < 2:
            print("Usage: /tool   ")
            proceed
            
          tool_name = parts[1]
          args = self._parse_prompt_arguments(parts[2:])
          await self.execute_tool(tool_name, args)
        elif command == '/prompts':
          await self.list_prompts()
        elif command == '/prompt':
          if len(parts) < 2:
            print("Usage: /prompt   ")
            proceed
          
          prompt_name = parts[1]
          args = self._parse_prompt_arguments(parts[2:])
          await self.execute_prompt(prompt_name, args)
        else:
          print(f"Unknown command: {command}")
        proceed
      
      # Process regular queries
      await self.process_query(query)
            
    except Exception as e:
      print(f"nError in chat loop: {e}")
      traceback.print_exc()

There are a bunch of helper functions to parse arguments and return the lists of accessible tools and prompts we registered earlier. Because it’s fairly straightforward, I won’t go into much detail here. You’ll be able to check the complete code in the event you have an interest.

As a substitute, let’s dive deeper into how the interactions between the MCP client and server work in several scenarios.

When working with resources, we use the self.sessions mapping to search out the suitable session (with a fallback option if needed) after which use that session to read the resource.

async def get_resource(self, resource_uri):
  """Retrieve and display content from an MCP resource."""
  session = self.sessions.get(resource_uri)
  
  # Fallback: find any session that handles this resource type
  if not session and resource_uri.startswith("changelog://"):
    session = next(
        (sess for uri, sess in self.sessions.items() 
         if uri.startswith("changelog://")), 
        None
    )
      
  if not session:
    print(f"Resource '{resource_uri}' not found.")
    return

  try:
    result = await session.read_resource(uri=resource_uri)
    if result and result.contents:
        print(f"nResource: {resource_uri}")
        print("Content:")
        print(result.contents[0].text)
    else:
        print("No content available.")
  except Exception as e:
    print(f"Error reading resource: {e}")
    traceback.print_exc()

To execute a tool, we follow an analogous process: start by finding the session after which use it to call the tool, passing its name and arguments.

async def execute_tool(self, tool_name, args):
  """Execute an MCP tool directly with given arguments."""
  session = self.sessions.get(tool_name)
  if not session:
      print(f"Tool '{tool_name}' not found.")
      return
  
  try:
      result = await session.call_tool(tool_name, arguments=args)
      print(f"nTool '{tool_name}' result:")
      print(result.content)
  except Exception as e:
      print(f"Error executing tool: {e}")
      traceback.print_exc()

No surprise here. The identical approach works for executing the prompt.

async def execute_prompt(self, prompt_name, args):
    """Execute an MCP prompt with given arguments and process the result."""
    session = self.sessions.get(prompt_name)
    if not session:
        print(f"Prompt '{prompt_name}' not found.")
        return
    
    try:
        result = await session.get_prompt(prompt_name, arguments=args)
        if result and result.messages:
            prompt_content = result.messages[0].content
            text = self._extract_prompt_text(prompt_content)
            
            print(f"nExecuting prompt '{prompt_name}'...")
            await self.process_query(text)
    except Exception as e:
        print(f"Error executing prompt: {e}")
        traceback.print_exc()

The one major use case we haven’t covered yet is handling a general, free-form input from a user (not one among specific commands). 
On this case, we send the initial request to the LLM first, then we parse the output, defining whether there are any tool calls. If tool calls are present, we execute them. Otherwise, we exit the infinite loop and return the reply to the user.

async def process_query(self, query):
  """Process a user query through Anthropic's Claude, handling tool calls iteratively."""
  messages = [{'role': 'user', 'content': query}]
  
  while True:
    response = self.anthropic.messages.create(
        max_tokens=2024,
        model='claude-3-7-sonnet-20250219', 
        tools=self.available_tools,
        messages=messages
    )
    
    assistant_content = []
    has_tool_use = False
    
    for content in response.content:
        if content.type == 'text':
            print(content.text)
            assistant_content.append(content)
        elif content.type == 'tool_use':
            has_tool_use = True
            assistant_content.append(content)
            messages.append({'role': 'assistant', 'content': assistant_content})
            
            # Execute the tool call
            session = self.sessions.get(content.name)
            if not session:
                print(f"Tool '{content.name}' not found.")
                break
                
            result = await session.call_tool(content.name, arguments=content.input)
            messages.append({
                "role": "user", 
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": content.id,
                    "content": result.content
                }]
            })
      
      if not has_tool_use:
          break

So, now we have now fully covered how the MCP chatbot actually works under the hood. Now, it’s time to check it in motion. You’ll be able to run it from the command line interface with the next command. 

python mcp_client_example_base.py

If you run the chatbot, you’ll first see the next introduction message outlining potential options:

MCP Chatbot Began!
Commands:
  quit                           - Exit the chatbot
  @periods                       - Show available changelog periods
  @                      - View changelog for specific period
  /tools                         - List available tools
  /tool       - Execute a tool with arguments
  /prompts                       - List available prompts
  /prompt     - Execute a prompt with arguments

From there, you possibly can check out different commands, for instance, 

• call the tool to list the databases available within the DB
• list all available prompts 
• use the prompt template, calling it like this /prompt sql_query_prompt query=”How many shoppers did now we have in May 2024?”. 

Finally, I can finish your chat by typing quit.

Query: /tool list_databases
[07/02/25 18:27:28] INFO     Processing request of type CallToolRequest                server.py:619
Tool 'list_databases' result:
[TextContent(type='text', text='INFORMATION_SCHEMAndatasetsndefaultnecommercenecommerce_dbninformation_schemansystemn', annotations=None, meta=None)]

Query: /prompts
Available prompts:
- sql_query_prompt: Create a SQL query prompt
  Arguments:
    - query

Query: /prompt sql_query_prompt query="How many shoppers did now we have in May 2024?"
[07/02/25 18:28:21] INFO     Processing request of type GetPromptRequest               server.py:619
Executing prompt 'sql_query_prompt'...
I'll create a SQL query to search out the number of consumers in May 2024.
[07/02/25 18:28:25] INFO     Processing request of type CallToolRequest                server.py:619
Based on the query results, here's the ultimate SQL query:
```sql
select uniqExact(user_id) as customer_count
from ecommerce.sessions
where toStartOfMonth(action_date) = '2024-05-01'
format TabSeparatedWithNames
```
Query: /tool execute_sql_query query="select uniqExact(user_id) as customer_count from ecommerce.sessions where toStartOfMonth(action_date) = '2024-05-01' format TabSeparatedWithNames"
I'll provide help to execute this SQL query to get the unique customer count for May 2024. Let me run this for you.
[07/02/25 18:30:09] INFO     Processing request of type CallToolRequest                server.py:619
The query has been executed successfully. The outcomes show that there have been 246,852 unique customers (unique user_ids) in May 2024 based on the ecommerce.sessions table.

Query: quit

Looks pretty cool! Our basic version is working well! Now, it’s time to take it one step further and make our chatbot smarter by teaching it to suggest relevant prompts on the fly based on customer input. 

Prompt suggestions

In practice, suggesting prompt templates that best match the user’s task will be incredibly helpful. Immediately, users of our chatbot have to either already find out about available prompts or at the least be curious enough to explore them on their very own to profit from what we’ve built. By adding a prompt suggestions feature, we are able to do that discovery for our users and make our chatbot significantly more convenient and user-friendly.

Let’s brainstorm ways so as to add this functionality. I’d approach this feature in the next way:

Evaluate the relevance of the prompts using the LLM. Iterate through all available prompt templates and, for each, assess whether the prompt is an excellent match for the user’s query.

Suggest an identical prompt to the user. If we found the relevant prompt template, share it with the user and ask whether or not they would really like to execute it. 

Merge the prompt template with the user input. If the user accepts, mix the chosen prompt with the unique query. Since prompt templates have placeholders, we would need the LLM to fill them in. Once we’ve merged the prompt template with the user’s query, we’ll have an updated message able to send to the LLM.

We’ll add this logic to the process_query function. Due to our modular design, it’s pretty easy so as to add this enhancement without disrupting the remainder of the code. 

Let’s start by implementing a function to search out probably the most relevant prompt template. We’ll use the LLM to judge each prompt and assign it a relevance rating from 0 to five. After that, we’ll filter out any prompts with a rating of two or lower and return only probably the most relevant one (the one with the very best relevance rating among the many remaining results).

async def _find_matching_prompt(self, query):
  """Find an identical prompt for the given query using LLM evaluation."""
  if not self.available_prompts:
    return None
  
  # Use LLM to judge prompt relevance
  prompt_scores = []
  
  for prompt in self.available_prompts:
    # Create evaluation prompt for the LLM
    evaluation_prompt = f"""
You might be an authority at evaluating whether a prompt template is relevant for a user query.

User Query: "{query}"

Prompt Template:
- Name: {prompt['name']}
- Description: {prompt['description']}

Rate the relevance of this prompt template for the user query on a scale of 0-5:
- 0: Completely irrelevant
- 1: Barely relevant
- 2: Somewhat relevant  
- 3: Moderately relevant
- 4: Highly relevant
- 5: Perfect match

Consider:
- Does the prompt template address the user's intent?
- Would using this prompt template provide a greater response than a generic query?
- Are the topics and context aligned?

Respond with only a single number (0-5) and no other text.
"""
      
    try:
      response = self.anthropic.messages.create(
          max_tokens=10,
          model='claude-3-7-sonnet-20250219',
          messages=[{'role': 'user', 'content': evaluation_prompt}]
      )
      
      # Extract the rating from the response
      score_text = response.content[0].text.strip()
      rating = int(score_text)
      
      if rating >= 3:  # Only consider prompts with rating >= 3
          prompt_scores.append((prompt, rating))
            
    except Exception as e:
        print(f"Error evaluating prompt {prompt['name']}: {e}")
        proceed
  
  # Return the prompt with the very best rating
  if prompt_scores:
      best_prompt, best_score = max(prompt_scores, key=lambda x: x[1])
      return best_prompt
  
  return None

The following function we’d like to implement is one that mixes the chosen prompt template with the user input. We’ll depend on the LLM to intelligently mix them, filling all placeholders as needed.

async def _combine_prompt_with_query(self, prompt_name, user_query):
  """Use LLM to mix prompt template with user query."""
  # First, get the prompt template content
  session = self.sessions.get(prompt_name)
  if not session:
      print(f"Prompt '{prompt_name}' not found.")
      return None
  
  try:
      # Find the prompt definition to get its arguments
      prompt_def = None
      for prompt in self.available_prompts:
          if prompt['name'] == prompt_name:
              prompt_def = prompt
              break
      
      # Prepare arguments for the prompt template
      args = {}
      if prompt_def and prompt_def.get('arguments'):
          for arg in prompt_def['arguments']:
              arg_name = arg.name if hasattr(arg, 'name') else arg.get('name', '')
              if arg_name:
                  # Use placeholder format for arguments
                  args[arg_name] = '<' + str(arg_name) + '>'
      
      # Get the prompt template with arguments
      result = await session.get_prompt(prompt_name, arguments=args)
      if not result or not result.messages:
          print(f"Couldn't retrieve prompt template for '{prompt_name}'")
          return None
      
      prompt_content = result.messages[0].content
      prompt_text = self._extract_prompt_text(prompt_content)
      
      # Create combination prompt for the LLM
      combination_prompt = f"""
You might be an authority at combining prompt templates with user queries to create optimized prompts.

Original User Query: "{user_query}"

Prompt Template:
{prompt_text}

Your task:
1. Analyze the user's query and the prompt template
2. Mix them intelligently to create a single, coherent prompt
3. Make sure the user's specific query/request is addressed throughout the context of the template
4. Maintain the structure and intent of the template while incorporating the user's query

Respond with only the combined prompt text, no explanations or additional text.
"""
      
      response = self.anthropic.messages.create(
          max_tokens=2048,
          model='claude-3-7-sonnet-20250219',
          messages=[{'role': 'user', 'content': combination_prompt}]
      )
      
      return response.content[0].text.strip()
      
  except Exception as e:
      print(f"Error combining prompt with query: {e}")
      return None

Then, we are going to simply update the process_query logic to envision for matching prompts, ask the user for confirmation and choose which message to send to the LLM.

async def process_query(self, query):
  """Process a user query through Anthropic's Claude, handling tool calls iteratively."""
  # Check if there's an identical prompt first
  matching_prompt = await self._find_matching_prompt(query)
  
  if matching_prompt:
    print(f"Found matching prompt: {matching_prompt['name']}")
    print(f"Description: {matching_prompt['description']}")
    
    # Ask user in the event that they need to use the prompt template
    use_prompt = input("Would you wish to use this prompt template? (y/n): ").strip().lower()
    
    if use_prompt == 'y' or use_prompt == 'yes':
        print("Combining prompt template along with your query...")
        
        # Use LLM to mix prompt template with user query
        combined_prompt = await self._combine_prompt_with_query(matching_prompt['name'], query)
        
        if combined_prompt:
            print(f"Combined prompt created. Processing...")
            # Process the combined prompt as a substitute of the unique query
            messages = [{'role': 'user', 'content': combined_prompt}]
        else:
            print("Did not mix prompt template. Using original query.")
            messages = [{'role': 'user', 'content': query}]
    else:
        # Use original query if user doesn't need to use the prompt
        messages = [{'role': 'user', 'content': query}]
  else:
    # Process the unique query if no matching prompt found
    messages = [{'role': 'user', 'content': query}]

  # print(messages)
  
  # Process the ultimate query (either original or combined)
  while True:
    response = self.anthropic.messages.create(
        max_tokens=2024,
        model='claude-3-7-sonnet-20250219', 
        tools=self.available_tools,
        messages=messages
    )
    
    assistant_content = []
    has_tool_use = False
    
    for content in response.content:
      if content.type == 'text':
          print(content.text)
          assistant_content.append(content)
      elif content.type == 'tool_use':
          has_tool_use = True
          assistant_content.append(content)
          messages.append({'role': 'assistant', 'content': assistant_content})
          
          # Log tool call information
          print(f"n[TOOL CALL] Tool: {content.name}")
          print(f"[TOOL CALL] Arguments: {json.dumps(content.input, indent=2)}")
          
          # Execute the tool call
          session = self.sessions.get(content.name)
          if not session:
              print(f"Tool '{content.name}' not found.")
              break
              
          result = await session.call_tool(content.name, arguments=content.input)
          
          # Log tool result
          print(f"[TOOL RESULT] Tool: {content.name}")
          print(f"[TOOL RESULT] Content: {result.content}")
          
          messages.append({
              "role": "user", 
              "content": [{
                  "type": "tool_result",
                  "tool_use_id": content.id,
                  "content": result.content
              }]
          })
      
    if not has_tool_use:
        break

Now, let’s test our updated version with a matter about our data. Excitingly, the chatbot was in a position to find the best prompt and use it to search out the best answer.

Query: How many shoppers did now we have in May 2024?
Found matching prompt: sql_query_prompt
Description: Create a SQL query prompt
Would you wish to use this prompt template? (y/n): y
Combining prompt template along with your query...
[07/05/25 14:38:58] INFO     Processing request of type GetPromptRequest               server.py:619
Combined prompt created. Processing...
I'll write a question to count unique customers who had sessions in May 2024. Since this can be a business metric, I'll exclude fraudulent sessions.

[TOOL CALL] Tool: execute_sql_query
[TOOL CALL] Arguments: {
  "query": "/* Count distinct users with non-fraudulent sessions in May 2024n   Using uniqExact for precise user countn   Filtering for May 2024 using toStartOfMonth and adding date range */nSELECT n    uniqExactIf(s.user_id, s.is_fraud = 0) AS active_customers_countnFROM ecommerce.sessions snWHERE toStartOfMonth(action_date) = toDate('2024-05-01')nFORMAT TabSeparatedWithNames"
}
[07/05/25 14:39:17] INFO     Processing request of type CallToolRequest                server.py:619
[TOOL RESULT] Tool: execute_sql_query
[TOOL RESULT] Content: [TextContent(type='text', text='active_customers_countn245287n', annotations=None, meta=None)]
The query shows we had 245,287 unique customers with legitimate (non-fraudulent) sessions in May 2024. Here's a breakdown of why I wrote the query this fashion:

1. Used uniqExactIf() to get precise count of unique users while excluding fraudulent sessions in a single step
2. Used toStartOfMonth() to make sure we capture all days in May 2024
3. Specified the date format properly with toDate('2024-05-01')
4. Used TabSeparatedWithNames format as required
5. Provided a meaningful column alias

Would you wish to see any variations of this evaluation, reminiscent of including fraudulent sessions or breaking down the numbers by country?

It’s all the time an excellent idea to check negative examples as well. On this case, the chatbot behaves as expected and doesn’t suggest an SQL-related prompt when given an unrelated query.

Query: How are you?
I should note that I'm an AI assistant focused on helping you're employed with the available tools, which include executing SQL queries, getting database/table information, and accessing GitHub PR data. I haven't got a tool specifically for responding to private questions.

I can provide help to:
- Query a ClickHouse database
- List databases and describe tables
- Get details about GitHub Pull Requests

What would you wish to find out about these areas?

Now that our chatbot is up and running, we’re able to wrap things up.

BONUS: quick and simple MCP client with smolagents

We’ve checked out low-level code that permits constructing highly customised MCP clients, but many use cases require only basic functionality. So, I made a decision to share with you a fast and easy implementation for scenarios whenever you need just the tools. We’ll use one among my favourite agent frameworks — smolagents from HuggingFace ().

# needed 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.loads(f.read())

os.environ["OPENAI_API_KEY"] = config['OPENAI_API_KEY']

# defining the LLM 
model = 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_"},
)

# prompt 
CLICKHOUSE_PROMPT_TEMPLATE = """
You might be a senior data analyst with greater than 10 years of experience writing complex SQL queries, specifically optimized for ClickHouse to reply user questions.

## Database Schema

You might be working with an e-commerce analytics database containing the next tables:

### Table: ecommerce.users 
**Description:** Customer information for the web shop
**Primary Key:** user_id
**Fields:** 
- user_id (Int64) - Unique customer identifier (e.g., 1000004, 3000004)
- country (String) - Customer's country of residence (e.g., "Netherlands", "United Kingdom")
- is_active (Int8) - Customer status: 1 = energetic, 0 = inactive
- age (Int32) - Customer age in full years (e.g., 31, 72)

### Table: ecommerce.sessions 
**Description:** User session data and transaction records
**Primary Key:** session_id
**Foreign Key:** user_id (references ecommerce.users.user_id)
**Fields:** 
- user_id (Int64) - Customer identifier linking to users table (e.g., 1000004, 3000004)
- session_id (Int64) - Unique session identifier (e.g., 106, 1023)
- action_date (Date) - Session start date (e.g., "2021-01-03", "2024-12-02")
- session_duration (Int32) - Session duration in seconds (e.g., 125, 49)
- os (String) - Operating system used (e.g., "Windows", "Android", "iOS", "MacOS")
- browser (String) - Browser used (e.g., "Chrome", "Safari", "Firefox", "Edge")
- is_fraud (Int8) - Fraud indicator: 1 = fraudulent session, 0 = legitimate
- revenue (Float64) - Purchase amount in USD (0.0 for non-purchase sessions, >0 for purchases)

## ClickHouse-Specific Guidelines

1. **Use ClickHouse-optimized functions:**
   - uniqExact() for precise unique counts
   - uniqExactIf() for conditional unique counts
   - quantile() functions for percentiles
   - Date functions: toStartOfMonth(), toStartOfYear(), today()

2. **Query formatting requirements:**
   - At all times end queries with "format TabSeparatedWithNames"
   - Use meaningful column aliases
   - Use proper JOIN syntax when combining tables
   - Wrap date literals in quotes (e.g., '2024-01-01')

3. **Performance considerations:**
   - Use appropriate WHERE clauses to filter data
   - Think about using HAVING for post-aggregation filtering
   - Use LIMIT when finding top/bottom results

4. **Data interpretation:**
   - revenue > 0 indicates a purchase order session
   - revenue = 0 indicates a browsing session without purchase
   - is_fraud = 1 sessions should typically be excluded from business metrics unless specifically analyzing fraud

## Response Format
Provide only the SQL query as your answer. Include transient reasoning in comments if the query logic is complex. 

## Examples

**Query:** How many shoppers made purchase in December 2024?
**Answer:** select uniqExact(user_id) as customers from ecommerce.sessions where toStartOfMonth(action_date) = '2024-12-01' and revenue > 0 format TabSeparatedWithNames

**Query:** What was the fraud rate in 2023, expressed as a percentage?
**Answer:** select 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.sessions where toStartOfYear(action_date) = '2023-01-01' format TabSeparatedWithNames

**Query:** What was the share of users using Windows yesterday?
**Answer:** select 100 * uniqExactIf(user_id, os = 'Windows') / uniqExact(user_id) as windows_share from ecommerce.sessions where action_date = today() - 1 format TabSeparatedWithNames

**Query:** What was the revenue from Dutch users aged 55 and older in December 2024?
**Answer:** select sum(s.revenue) as total_revenue from ecommerce.sessions as s inner join ecommerce.users as u on s.user_id = u.user_id where u.country = 'Netherlands' and u.age >= 55 and toStartOfMonth(s.action_date) = '2024-12-01' format TabSeparatedWithNames

**Query:** What are the median and interquartile range (IQR) of purchase revenue for every country?
**Answer:** select country, median(revenue) as median_revenue, quantile(0.25)(revenue) as q25_revenue, quantile(0.75)(revenue) as q75_revenue from ecommerce.sessions as s inner join ecommerce.users as u on u.user_id = s.user_id where revenue > 0 group by country format TabSeparatedWithNames

**Query:** What's the typical variety of days between the primary session and the primary purchase for users who made at the least one purchase?
**Answer:** select avg(first_purchase - first_action_date) as avg_days_to_purchase from (select user_id, min(action_date) as first_action_date, minIf(action_date, revenue > 0) as first_purchase, max(revenue) as max_revenue from ecommerce.sessions group by user_id) where max_revenue > 0 format TabSeparatedWithNames

**Query:** What's the variety of sessions in December 2024, broken down by operating systems, including the totals?
**Answer:** select os, uniqExact(session_id) as session_count from ecommerce.sessions where toStartOfMonth(action_date) = '2024-12-01' group by os with totals format TabSeparatedWithNames

**Query:** Do now we have customers who used multiple browsers during 2024? If that's the case, please calculate the number of consumers for every combination of browsers.
**Answer:** select browsers, count(*) as customer_count from (select user_id, arrayStringConcat(arraySort(groupArray(distinct browser)), ', ') as browsers from ecommerce.sessions where 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 users?
**Answer:** select browser, 100 * uniqExactIf(user_id, is_fraud = 1) / uniqExact(user_id) as fraud_rate from ecommerce.sessions group by browser order by fraud_rate desc limit 1 format TabSeparatedWithNames

**Query:** Which country had the very best variety of first-time users in 2024?
**Answer:** select country, count(distinct user_id) as new_users from (select user_id, min(action_date) as first_date from ecommerce.sessions group by user_id having toStartOfYear(first_date) = '2024-01-01') as t inner join ecommerce.users as u on t.user_id = u.user_id group by country order by new_users desc limit 1 format TabSeparatedWithNames

---

**Your Task:** Using all of the provided information above, write a ClickHouse SQL query to reply the next customer query: 
{query}
"""

with ToolCollection.from_mcp(server_parameters, trust_remote_code=True) as tool_collection:
  agent = ToolCallingAgent(tools=[*tool_collection.tools], model=model)
  prompt = CLICKHOUSE_PROMPT_TEMPLATE.format(
      query = 'How many shoppers did now we have in May 2024?'
  )
  response = agent.run(prompt)

In consequence, we received the proper answer.

Should you don’t need much customisation or integration with prompts and resources, this implementation is unquestionably the strategy to go.

Summary

In this text, we built a chatbot that integrates with MCP servers and leverages all the advantages of standardisation to access tools, prompts, and resources seamlessly.

We began with a basic implementation able to listing and accessing MCP capabilities. Then, we enhanced our chatbot with a wise feature that implies relevant prompt templates to users based on their input. This makes our product more intuitive and user-friendly, especially for users unfamiliar with the whole library of accessible prompts.

To implement our chatbot, we used relatively low-level code, supplying you with a greater understanding of how the MCP protocol works under the hood and what happens whenever you use AI tools like Claude Desktop or Cursor.

As a bonus, we also discussed the smolagents implementation that enables you to quickly deploy an MCP client integrated with tools.

Reference

This text is inspired by the short course from