Erstellen eines сustom MCP Chatbot | Auf Information Science

Eine Methode zur Standardisierung der Kommunikation zwischen AI -Anwendungen und externen Instruments oder Datenquellen. Diese Standardisierung hilft, die Anzahl der erforderlichen Integrationen zu verringern (von n*m bis n+m):

• Sie können mit Group gebaute MCP-Server verwenden, wenn Sie gemeinsame Funktionen benötigen, Zeit sparen und die Notwendigkeit vermeiden, das Rad jedes Mal neu zu erfinden.
• Sie können auch Ihre eigenen Instruments und Ressourcen aufdecken und sie für andere zur Verfügung stellen.

In Mein vorheriger ArtikelWir haben die Analytics Toolbox erstellt (eine Sammlung von Instruments, die Ihre tägliche Routine automatisieren können). Wir haben einen MCP -Server erstellt und seine Funktionen mit vorhandenen Shoppers wie MCP Inspector oder Claude Desktop verwendet.

Jetzt möchten wir diese Instruments direkt in unseren KI -Anwendungen verwenden. Lassen Sie uns dazu unseren eigenen MCP -Kunden aufbauen. Wir werden einen ziemlich niedrigen Code schreiben, der Ihnen auch ein klareres Bild davon gibt, wie Instruments wie Claude Code mit MCP unter der Motorhaube interagieren.

Darüber hinaus möchte ich die derzeit befindliche Funktion implementieren ((Juli 2025) Fehlen von Claude Desktop: Die Fähigkeit des LLM, automatisch zu prüfen, ob es eine geeignete Eingabeaufforderung für die anstehende Aufgabe enthält, und sie verwenden. Im Second müssen Sie die Vorlage manuell auswählen, was nicht sehr bequem ist.

Als Bonus werde ich auch eine hochrangige Implementierung mit dem Smolagents-Framework teilen, das best für Szenarien ist, wenn Sie nur mit MCP-Instruments arbeiten und nicht viel Anpassung benötigen.

MCP -Protokollübersicht

Hier ist eine kurze Zusammenfassung des MCP, um sicherzustellen, dass wir auf derselben Seite sind. MCP ist ein von Anthropic entwickeltes Protokoll, um die Artwork und Weise zu standardisieren, wie LLMs mit der Außenwelt interagieren.

Es folgt einer Consumer-Server-Architektur und besteht aus drei Hauptkomponenten:

• Gastgeber Ist die vom Benutzer ausgerichtete Anwendung.
• MCP -Consumer ist eine Komponente innerhalb des Hosts, die eine Eins-zu-Eins-Verbindung zum Server herstellt und mithilfe von Nachrichten, die vom MCP-Protokoll definiert wurden, kommuniziert.
• MCP -Server Enthält Funktionen wie schnelle Vorlagen, Ressourcen und Instruments.

Da haben wir schon implementierte den MCP -Server Zuvor konzentrieren wir uns diesmal auf den Aufbau des MCP -Kunden. Wir beginnen mit einer relativ einfachen Implementierung und fügen später die Möglichkeit hinzu, im laufenden Fliegen Eingangsvorlagen dynamisch auszuwählen.

Sie finden den vollständigen Code auf Github.

Erstellen des MCP -Chatbots

Beginnen wir mit dem ersten Setup: Wir laden die anthropische API -Style aus einer Konfigurationsdatei und passen Python an asyncio Ereignisschleife zur Unterstützung verschachtelter Veranstaltungsschleifen.

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

nest_asyncio.apply()

Beginnen wir zunächst ein Skelett unseres Programms, um ein klares Bild der hochrangigen Architektur der Anwendung zu machen.

async def fundamental():
    """Primary 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(fundamental())

Wir beginnen zunächst eine Instanz der MCP_ChatBot Klasse. Der Chatbot startet mit der Erkennung der verfügbaren MCP -Funktionen (Iteration aller konfigurierten MCP -Server, Aufbau von Verbindungen und Anfragen ihrer Funktionslisten).

Sobald Verbindungen eingerichtet sind, werden wir eine unendliche Schleife initialisieren, in der der Chatbot die Benutzeranfragen hört, bei Bedarf Instruments aufruft und diesen Zyklus fortsetzt, bis der Prozess manuell gestoppt wird.

Schließlich werden wir einen Reinigungsschritt durchführen, um alle offenen Verbindungen zu schließen.

Gehen wir nun detaillierter durch jede Stufe.

Initialisierung der Chatbot -Klasse

Beginnen wir mit der Erstellung der Klasse und definieren die __init__ Verfahren. Die Hauptfelder der Chatbot -Klasse sind:

• exit_stack Verwaltet den Lebenszyklus mehrerer asynchronisierter Threads (Verbindungen zu MCP -Servern), um sicherzustellen, dass alle Verbindungen angemessen geschlossen werden, selbst wenn wir während der Ausführung einen Fehler haben. Diese Logik wird in der implementiert cleanup Funktion.
• anthropic ist ein Consumer für anthropische API, mit dem Nachrichten an LLM gesendet werden.
• available_tools Und available_prompts sind die Pay attention von Instruments und Eingabeaufforderungen, mit denen alle MCP -Server, mit denen wir verbunden sind, aufgedeckt werden.
• periods ist eine Zuordnung von Instruments, Aufforderungen und Ressourcen zu ihren jeweiligen MCP -Sitzungen. Auf diese Weise kann der Chatbot Anforderungen an den richtigen MCP -Server weiterleiten, wenn das LLM ein bestimmtes Device auswählt.

class MCP_ChatBot:
  """
  MCP (Mannequin Context Protocol) ChatBot that connects to a number of MCP servers
  and gives a conversational interface utilizing Anthropic's Claude.
    
  Helps instruments, prompts, and assets from related MCP servers.
  """
    
  def __init__(self):
    self.exit_stack = AsyncExitStack() 
    self.anthropic = Anthropic() # Consumer for Anthropic API
    self.available_tools = () # Instruments from all related servers
    self.available_prompts = () # Prompts from all related servers  
    self.periods = {} # Maps instrument/immediate/useful resource names to MCP periods

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

Verbindung zu Servern herstellen

Die erste Aufgabe für unseren Chatbot besteht darin, Verbindungen mit allen konfigurierten MCP -Servern zu initiieren und zu ermitteln, welche Funktionen wir verwenden können.

Die Liste der MCP -Server, mit denen unser Agent eine Verbindung herstellen kann server_config.json Datei. Ich habe Verbindungen mit drei MCP -Servern aufgestellt:

• Analyst_toolkit Ist meine Implementierung der alltäglichen analytischen Instruments, die wir im vorherigen Artikel diskutiert haben,
• Dateisystem ermöglicht dem Agenten, mit Dateien zu arbeiten,
• Bringen Hilft LLMs beim Abrufen des Inhalts von Webseiten und konvertieren sie von HTML in Markdown, um eine bessere Lesbarkeit zu erhalten.

{
  "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",
        "/Customers/marie/Desktop",
        "/Customers/marie/Paperwork/github"
      )
    },
    "fetch": {
        "command": "uvx",
        "args": ("mcp-server-fetch")
      }
  }
}

Zuerst lesen wir die Konfigurationsdatei, analysieren sie und stellen dann eine Verbindung zu jedem aufgeführten Server her.

async def connect_to_servers(self):
  """Load server configuration and hook up with all configured MCP servers."""
  strive:
    with open("server_config.json", "r") as file:
      information = json.load(file)
    
    servers = information.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()
    elevate

Für jeden Server führen wir mehrere Schritte aus, um die Verbindung herzustellen:

• Auf Transportebene, Wir Starten Sie den MCP -Server als STDIO -Prozess und erhalten Sie Streams zum Senden und Empfangen von Nachrichten.
• Auf Sitzungsebenewir erstellen eine ClientSession Einbeziehung der Streams und dann führen wir den MCP -Handshake durch Anrufe durch initialize Verfahren.
• Wir haben sowohl die Sitzungs- als auch die Transportobjekte im Kontextmanager registriert exit_stack Um sicherzustellen, dass alle Verbindungen am Ende ordnungsgemäß geschlossen werden.
• Der letzte Schritt ist zu Serverfunktionen registrieren. Wir haben diese Funktionalität in eine separate Funktion eingepackt und werden sie in Kürze diskutieren.

async def connect_to_server(self, server_name, server_config):
    """Connect 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()

Durch die Registrierung von Funktionen werden alle Instruments, Eingabeaufforderungen und Ressourcen aus der Sitzung iteriert. Infolgedessen aktualisieren wir die internen Variablen periods (Zuordnung zwischen Ressourcen und einer bestimmten Sitzung zwischen dem MCP -Consumer und dem Server), available_prompts Und available_tools.

async def _register_server_capabilities(self, session, server_name):
  """Register instruments, prompts and assets from a single server."""
  capabilities = (
    ("instruments", session.list_tools, self._register_tools),
    ("prompts", session.list_prompts, self._register_prompts), 
    ("assets", 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 instrument in response.instruments:
    self.periods(instrument.identify) = session
    self.available_tools.append({
        "identify": instrument.identify,
        "description": instrument.description,
        "input_schema": instrument.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(immediate.identify) = session
        self.available_prompts.append({
            "identify": immediate.identify,
            "description": immediate.description,
            "arguments": immediate.arguments
        })

async def _register_resources(self, response, session):
  """Register assets from server response."""
  if response and response.assets:
    for useful resource in response.assets:
        resource_uri = str(useful resource.uri)
        self.periods(resource_uri) = session

Am Ende dieser Part unsere, unsere MCP_ChatBot Das Objekt hat alles, was es braucht, um mit den Benutzern zu interagieren:

• Verbindungen zu allen konfigurierten MCP -Servern werden hergestellt.
• Alle Eingabeaufforderungen, Ressourcen und Instruments sind registriert, einschließlich Beschreibungen, die für LLM erforderlich sind, um zu verstehen, wie diese Funktionen verwendet werden.
• Zuordnungen zwischen diesen Ressourcen und ihren jeweiligen Sitzungen werden gespeichert, sodass wir genau wissen, wohin wir jede Anfrage senden sollen.

Chat -Schleife

Es ist additionally Zeit, unseren Chat mit den Benutzern zu beginnen, indem Sie die erstellen chat_loop Funktion.

Wir werden zuerst alle verfügbaren Befehle mit dem Benutzer weitergeben:

• Auflistung von Ressourcen, Instruments und Eingabeaufforderungen
• Ausführung eines Device -Anrufs
• Ressource anzeigen
• Verwenden einer Eingabeaufforderung Vorlage
• den Chat beenden (Es ist wichtig, einen klaren Weg zu haben, um die unendliche Schleife zu verlassen).

Danach werden wir eine unendliche Schleife eingeben, in der wir basierend auf den Benutzereingaben die entsprechende Aktion ausführen: unabhängig davon, ob es sich um einen der obigen Befehle handelt oder eine Anfrage an die LLM.

async def chat_loop(self):
  """Primary interactive chat loop with command processing."""
  print("nMCP Chatbot Began!")
  print("Instructions:")
  print("  stop                           - Exit the chatbot")
  print("  @intervals                       - Present obtainable changelog intervals") 
  print("  @<interval>                      - View changelog for particular interval")
  print("  /instruments                         - Listing obtainable instruments")
  print("  /instrument <identify> <arg1=value1>     - Execute a instrument with arguments")
  print("  /prompts                       - Listing obtainable prompts")
  print("  /immediate <identify> <arg1=value1>   - Execute a immediate with arguments")
  
  whereas True:
    strive:
      question = enter("nQuery: ").strip()
      if not question:
          proceed

      if question.decrease() == 'stop':
          break
      
      # Deal with useful resource requests (@command)
      if question.startswith('@'):
        interval = question(1:)
        resource_uri = "changelog://intervals" if interval == "intervals" else f"changelog://{interval}"
        await self.get_resource(resource_uri)
        proceed
      
      # Deal with slash instructions
      if question.startswith('/'):
        elements = self._parse_command_arguments(question)
        if not elements:
          proceed
            
        command = elements(0).decrease()
        
        if command == '/instruments':
          await self.list_tools()
        elif command == '/instrument':
          if len(elements) < 2:
            print("Utilization: /instrument <identify> <arg1=value1> <arg2=value2>")
            proceed
            
          tool_name = elements(1)
          args = self._parse_prompt_arguments(elements(2:))
          await self.execute_tool(tool_name, args)
        elif command == '/prompts':
          await self.list_prompts()
        elif command == '/immediate':
          if len(elements) < 2:
            print("Utilization: /immediate <identify> <arg1=value1> <arg2=value2>")
            proceed
          
          prompt_name = elements(1)
          args = self._parse_prompt_arguments(elements(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()

Es gibt eine Reihe von Helferfunktionen, um Argumente zu analysieren und die Pay attention der verfügbaren Instruments und Eingabeaufforderungen zurückzugeben, die wir zuvor registriert haben. Da es ziemlich einfach ist, werde ich hier nicht viel Particulars eingehen. Sie können überprüfen der vollständige Code Wenn Sie interessiert sind.

Lassen Sie uns stattdessen tiefer in die Artwork und Weise eintauchen, wie die Interaktionen zwischen dem MCP -Consumer und dem Server in verschiedenen Szenarien funktionieren.

Bei der Arbeit mit Ressourcen verwenden wir die self.periods Mapping, um die entsprechende Sitzung zu finden (mit einer Fallback -Possibility bei Bedarf) und dann diese Sitzung zum Lesen der Ressource.

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 sort
  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 obtainable.")
  besides Exception as e:
    print(f"Error studying useful resource: {e}")
    traceback.print_exc()

Um ein Device auszuführen, folgen wir einem ähnlichen Vorgang: Beginnen Sie zunächst die Sitzung und verwenden Sie es dann, um das Device aufzurufen und seinen Namen und seine Argumente zu übergeben.

async def execute_tool(self, tool_name, args):
  """Execute an MCP instrument immediately with given arguments."""
  session = self.periods.get(tool_name)
  if not session:
      print(f"Device '{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 instrument: {e}")
      traceback.print_exc()

Kein Wunder hier. Der gleiche Ansatz eignet sich zur Ausführung der Eingabeaufforderung.

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()

Der einzige wichtige Anwendungsfall, den wir noch nicht behandelt haben, ist die Behandlung einer allgemeinen Freiform-Eingabe eines Benutzers (nicht eines der spezifischen Befehle).
In diesem Fall senden wir zuerst die anfängliche Anfrage an die LLM, dann analysieren wir die Ausgabe und definieren, ob es Toolaufrufe gibt. Wenn Device -Anrufe vorhanden sind, führen wir sie aus. Andernfalls beenden wir die unendliche Schleife und geben die Antwort an den Benutzer zurück.

async def process_query(self, question):
  """Course of a consumer question by way of Anthropic's Claude, dealing with instrument calls iteratively."""
  messages = ({'position': 'consumer', 'content material': question})
  
  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.sort == 'textual content':
            print(content material.textual content)
            assistant_content.append(content material)
        elif content material.sort == 'tool_use':
            has_tool_use = True
            assistant_content.append(content material)
            messages.append({'position': 'assistant', 'content material': assistant_content})
            
            # Execute the instrument name
            session = self.periods.get(content material.identify)
            if not session:
                print(f"Device '{content material.identify}' not discovered.")
                break
                
            end result = await session.call_tool(content material.identify, arguments=content material.enter)
            messages.append({
                "position": "consumer", 
                "content material": ({
                    "sort": "tool_result",
                    "tool_use_id": content material.id,
                    "content material": end result.content material
                })
            })
      
      if not has_tool_use:
          break

Wir haben jetzt voll bedeckt, wie der MCP -Chatbot tatsächlich unter der Motorhaube funktioniert. Jetzt ist es Zeit, es in Aktion zu testen. Sie können es mit dem folgenden Befehl über die Befehlszeilenschnittstelle ausführen.

python mcp_client_example_base.py

Wenn Sie den Chatbot ausführen, sehen Sie zunächst die folgende Einführungsnachricht, in denen potenzielle Optionen beschrieben werden:

MCP Chatbot Began!
Instructions:
  stop                           - Exit the chatbot
  @intervals                       - Present obtainable changelog intervals
  @<interval>                      - View changelog for particular interval
  /instruments                         - Listing obtainable instruments
  /instrument <identify> <arg1=value1>     - Execute a instrument with arguments
  /prompts                       - Listing obtainable prompts
  /immediate <identify> <arg1=value1>   - Execute a immediate with arguments

Von dort aus können Sie zum Beispiel verschiedene Befehle ausprobieren,

• Rufen Sie das Device an, um die im DB verfügbaren Datenbanken aufzulisten
• Pay attention Sie alle verfügbaren Eingabeaufforderungen auf
• Verwenden Sie die Eingabeaufforderung Vorlage und rufen Sie sie so auf /immediate sql_query_prompt query=”What number of prospects did now we have in Could 2024?”.

Schließlich kann ich Ihren Chat durch Tippen beenden stop.

Question: /instrument list_databases
(07/02/25 18:27:28) INFO     Processing request of sort CallToolRequest                server.py:619
Device 'list_databases' end result:
(TextContent(sort='textual content', textual content='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 prospects did now we have in Could 2024?"
(07/02/25 18:28:21) INFO     Processing request of sort GetPromptRequest               server.py:619
Executing immediate 'sql_query_prompt'...
I am going to create a SQL question to seek out the variety of prospects in Could 2024.
(07/02/25 18:28:25) INFO     Processing request of sort 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: /instrument 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 am going to assist you execute this SQL question to get the distinctive buyer depend for Could 2024. Let me run this for you.
(07/02/25 18:30:09) INFO     Processing request of sort CallToolRequest                server.py:619
The question has been executed efficiently. The outcomes present that there have been 246,852 distinctive prospects (distinctive user_ids) in Could 2024 primarily based on the ecommerce.periods desk.

Question: stop

Sieht ziemlich cool aus! Unsere Grundversion funktioniert intestine! Jetzt ist es Zeit, einen Schritt weiter zu gehen und unseren Chatbot intelligenter zu gestalten, indem wir es unterrichten, um relevante Eingabeaufforderungen im laufenden Fliegen auf der Grundlage von Kundeneingaben vorzuschlagen.

Auffordern Vorschläge

In der Praxis kann es unglaublich hilfreich sein, schnelle Vorlagen vorzuschlagen, die am besten mit der Aufgabe des Benutzers übereinstimmen. Derzeit müssen Benutzer unseres Chatbot entweder über verfügbare Eingabeaufforderungen erfahren oder zumindest neugierig genug sein, um sie selbst zu erkunden, um von dem zu profitieren, was wir aufgebaut haben. Durch das Hinzufügen einer schnitten Vorschläge können wir diese Entdeckung für unsere Benutzer durchführen und unseren Chatbot erheblich bequemer und benutzerfreundlicher machen.

Lassen Sie uns Brainstorming -Möglichkeiten, um diese Funktionalität hinzuzufügen. Ich würde diese Funktion auf folgende Weise nähern:

Bewerten Sie die Relevanz der Eingabeaufforderungen anhand des LLM. Idieren Sie alle verfügbaren Eingabeaufentwicklungsvorlagen durch und beurteilen Sie für jeden, ob die Eingabeaufforderung intestine mit der Abfrage des Benutzers übereinstimmt.

Schlagen Sie dem Benutzer eine passende Eingabeaufforderung vor. Wenn wir die relevante Eingabeaufforderungsvorlage finden, teilen Sie sie mit dem Benutzer und fragen Sie, ob sie sie ausführen möchten.

Führen Sie die Eingabeaufforderung Vorlage mit der Benutzereingabe zusammen. Wenn der Benutzer akzeptiert, kombinieren Sie die ausgewählte Eingabeaufforderung mit der ursprünglichen Abfrage. Da schnelle Vorlagen Platzhalter haben, benötigen wir möglicherweise die LLM, um sie auszufüllen. Sobald wir die Eingabeaufforderung Vorlage mit der Abfrage des Benutzers zusammengefasst haben, haben wir eine aktualisierte Nachricht, die an die LLM gesendet werden kann.

Wir werden diese Logik zum Fügen process_query Funktion. Dank unseres modularen Designs ist es ziemlich einfach, diese Verbesserung hinzuzufügen, ohne den Relaxation des Codes zu stören.

Beginnen wir mit der Implementierung einer Funktion, um die relevanteste Eingabeaufforderungsvorlage zu finden. Wir werden das LLM verwenden, um jede Eingabeaufforderung zu bewerten und ihm eine Relevanzbewertung von 0 bis 5 zuzuweisen. Danach werden wir alle Eingaben mit einer Punktzahl von 2 oder niedriger herausfiltern und nur die relevanteste zurückgeben (die mit der höchsten Relevanzbewertung unter den verbleibenden Ergebnissen).

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 judge immediate relevance
  prompt_scores = ()
  
  for immediate in self.available_prompts:
    # Create analysis immediate for the LLM
    evaluation_prompt = f"""
You're an knowledgeable at evaluating whether or not a immediate template is related for a consumer question.

Consumer Question: "{question}"

Immediate Template:
- Identify: {immediate('identify')}
- Description: {immediate('description')}

Charge 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: Good match

Take into account:
- Does the immediate template tackle 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=({'position': 'consumer', 'content material': 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 think about prompts with rating >= 3
          prompt_scores.append((immediate, rating))
            
    besides Exception as e:
        print(f"Error evaluating immediate {immediate('identify')}: {e}")
        proceed
  
  # Return the immediate with the best rating
  if prompt_scores:
      best_prompt, best_score = max(prompt_scores, key=lambda x: x(1))
      return best_prompt
  
  return None

Die nächste Funktion, die wir implementieren müssen, ist eine, die die ausgewählte Eingabeaufforderung Vorlage mit der Benutzereingabe kombiniert. Wir werden uns auf die LLM verlassen, um sie clever zu kombinieren und alle Platzhalter bei Bedarf zu füllen.

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('identify') == 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.identify if hasattr(arg, 'identify') else arg.get('identify', '')
              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're an knowledgeable at combining immediate templates with consumer queries to create optimized prompts.

Authentic Consumer Question: "{user_query}"

Immediate Template:
{prompt_text}

Your job:
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 throughout 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 further textual content.
"""
      
      response = self.anthropic.messages.create(
          max_tokens=2048,
          mannequin='claude-3-7-sonnet-20250219',
          messages=({'position': 'consumer', 'content material': combination_prompt})
      )
      
      return response.content material(0).textual content.strip()
      
  besides Exception as e:
      print(f"Error combining immediate with question: {e}")
      return None

Dann werden wir einfach die aktualisieren process_query Logik Um nach Übereinstimmungsanforderungen zu prüfen, fragen Sie den Benutzer um eine Bestätigung und entscheiden Sie, welche Nachricht an die LLM gesendet werden soll.

async def process_query(self, question):
  """Course of a consumer question by way of Anthropic's Claude, dealing with instrument calls iteratively."""
  # Verify 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('identify')}")
    print(f"Description: {matching_prompt('description')}")
    
    # Ask consumer in the event that they need 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 along with your question...")
        
        # Use LLM to mix immediate template with consumer question
        combined_prompt = await self._combine_prompt_with_query(matching_prompt('identify'), question)
        
        if combined_prompt:
            print(f"Mixed immediate created. Processing...")
            # Course of the mixed immediate as an alternative of the unique question
            messages = ({'position': 'consumer', 'content material': combined_prompt})
        else:
            print("Failed to mix immediate template. Utilizing unique question.")
            messages = ({'position': 'consumer', 'content material': question})
    else:
        # Use unique question if consumer does not need to use the immediate
        messages = ({'position': 'consumer', 'content material': question})
  else:
    # Course of the unique question if no matching immediate discovered
    messages = ({'position': 'consumer', 'content material': question})

  # 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.sort == 'textual content':
          print(content material.textual content)
          assistant_content.append(content material)
      elif content material.sort == 'tool_use':
          has_tool_use = True
          assistant_content.append(content material)
          messages.append({'position': 'assistant', 'content material': assistant_content})
          
          # Log instrument name info
          print(f"n(TOOL CALL) Device: {content material.identify}")
          print(f"(TOOL CALL) Arguments: {json.dumps(content material.enter, indent=2)}")
          
          # Execute the instrument name
          session = self.periods.get(content material.identify)
          if not session:
              print(f"Device '{content material.identify}' not discovered.")
              break
              
          end result = await session.call_tool(content material.identify, arguments=content material.enter)
          
          # Log instrument end result
          print(f"(TOOL RESULT) Device: {content material.identify}")
          print(f"(TOOL RESULT) Content material: {end result.content material}")
          
          messages.append({
              "position": "consumer", 
              "content material": ({
                  "sort": "tool_result",
                  "tool_use_id": content material.id,
                  "content material": end result.content material
              })
          })
      
    if not has_tool_use:
        break

Lassen Sie uns nun unsere aktualisierte Model mit einer Frage zu unseren Daten testen. Aufregend conflict der Chatbot in der Lage, die richtige Eingabeaufforderung zu finden und sie zu verwenden, um die richtige Antwort zu finden.

Question: What number of prospects did now we have in Could 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 along with your question...
(07/05/25 14:38:58) INFO     Processing request of sort GetPromptRequest               server.py:619
Mixed immediate created. Processing...
I am going to write a question to depend distinctive prospects who had periods in Could 2024. Since it is a enterprise metric, I am going to exclude fraudulent periods.

(TOOL CALL) Device: execute_sql_query
(TOOL CALL) Arguments: {
  "question": "/* Rely distinct customers with non-fraudulent periods in Could 2024n   Utilizing uniqExact for exact consumer countn   Filtering for Could 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 sort CallToolRequest                server.py:619
(TOOL RESULT) Device: execute_sql_query
(TOOL RESULT) Content material: (TextContent(sort='textual content', textual content='active_customers_countn245287n', annotations=None, meta=None))
The question reveals we had 245,287 distinctive prospects with professional (non-fraudulent) periods in Could 2024. This is a breakdown of why I wrote the question this manner:

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 Could 2024
3. Specified the date format correctly with toDate('2024-05-01')
4. Used TabSeparatedWithNames format as required
5. Supplied a significant column alias

Would you prefer to see any variations of this evaluation, equivalent to together with fraudulent periods or breaking down the numbers by nation?

Es ist immer eine gute Idee, damaging Beispiele zu testen. In diesem Fall verhält sich der Chatbot wie erwartet und schlägt keine SQL-bezogene Eingabeaufforderung vor, wenn sie eine nicht verwandte Frage bezieht.

Question: How are you?
I ought to notice that I am an AI assistant targeted on serving to you're employed with the obtainable instruments, which embrace executing SQL queries, getting database/desk info, and accessing GitHub PR information. I haven't got a instrument particularly for responding to private questions.

I may help you:
- Question a ClickHouse database
- Listing databases and describe tables
- Get details about GitHub Pull Requests

What would you prefer to find out about these areas?

Jetzt, da unser Chatbot in Betrieb ist, sind wir bereit, Dinge abzuschließen.

Bonus: Schneller und einfacher MCP -Consumer mit Smolagents

Wir haben uns mit einem Code mit niedrigem Niveau befasst, mit dem hoch angepasste MCP-Shoppers erstellt werden können. Viele Anwendungsfälle erfordern jedoch nur grundlegende Funktionen. Additionally habe ich beschlossen, Ihnen eine schnelle und unkomplizierte Implementierung für Szenarien mitzuteilen, wenn Sie nur die Instruments benötigen. Wir werden eines meiner Lieblings -Agent -Frameworks verwenden – Smolagents von Huggingface (Ich habe diesen Framework im Element in diskutiert Mein vorheriger Artikel).

# 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.hundreds(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 information analyst with greater than 10 years of expertise writing advanced SQL queries, particularly optimized for ClickHouse to reply consumer questions.

## Database Schema

You're working with an e-commerce analytics database containing the next tables:

### Desk: ecommerce.customers 
**Description:** Buyer info for the web store
**Major 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 information and transaction information
**Major 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 length 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 = professional
- income (Float64) - Buy quantity in USD (0.0 for non-purchase periods, >0 for purchases)

## ClickHouse-Particular Tips

1. **Use ClickHouse-optimized features:**
   - uniqExact() for exact distinctive counts
   - uniqExactIf() for conditional distinctive counts
   - quantile() features for percentiles
   - Date features: toStartOfMonth(), toStartOfYear(), at this time()

2. **Question formatting necessities:**
   - All the time 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 issues:**
   - Use applicable WHERE clauses to filter information
   - Think about using HAVING for post-aggregation filtering
   - Use LIMIT when discovering prime/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 advanced. 

## Examples

**Query:** What number of prospects made buy in December 2024?
**Reply:** choose uniqExact(user_id) as prospects from ecommerce.periods the place toStartOfMonth(action_date) = '2024-12-01' and income > 0 format TabSeparatedWithNames

**Query:** What was the fraud fee 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 = at this time() - 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 inside 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 inside 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 at the least 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 techniques, 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 prospects who used a number of browsers throughout 2024? In that case, please calculate the variety of prospects 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 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 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 inside 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 Process:** Utilizing all of the supplied info 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.instruments), mannequin=mannequin)
  immediate = CLICKHOUSE_PROMPT_TEMPLATE.format(
      query = 'What number of prospects did now we have in Could 2024?'
  )
  response = agent.run(immediate)

Infolgedessen haben wir die richtige Antwort erhalten.

Wenn Sie nicht viel Anpassung oder Integration mit Eingabeaufforderungen und Ressourcen benötigen, ist diese Implementierung definitiv der richtige Weg.

Zusammenfassung

In diesem Artikel haben wir einen Chatbot erstellt, der sich in MCP -Server integriert und alle Vorteile der Standardisierung für den Zugriff auf Instruments, Eingabeaufforderungen und Ressourcen nahtlos nutzt.

Wir haben mit einer grundlegenden Implementierung begonnen, die in der Lage ist, MCP -Funktionen aufzulisten und zugreifen zu können. Anschließend haben wir unseren Chatbot mit einer intelligenten Funktion erweitert, die den Benutzern basierend auf deren Eingaben relevante Eingabeaufforderungsvorlagen vorschlägt. Dies macht unser Produkt intuitiver und benutzerfreundlicher, insbesondere für Benutzer, die mit der vollständigen Bibliothek verfügbarer Eingabeaufforderungen nicht vertraut sind.

Um unseren Chatbot zu implementieren, haben wir einen relativ niedrigen Code verwendet, um zu verstehen, wie das MCP-Protokoll unter der Motorhaube funktioniert und was passiert, wenn Sie KI-Instruments wie Claude Desktop oder Cursor verwenden.

Als Bonus haben wir auch die Smolagents -Implementierung besprochen, mit der Sie schnell einen MCP -Consumer bereitstellen können, der in Instruments integriert ist.

Danke fürs Lesen. Ich hoffe, dieser Artikel conflict aufschlussreich. Denken Sie daran, Einsteins Rat: „Das Wichtigste ist nicht, die Befragung aufzuhören. Neugier hat ihren eigenen Grund für das Bestehen.“ Möge Ihre Neugier Sie zu Ihrem nächsten großen Einblick führen.

Referenz

Dieser Artikel ist von der inspiriert “MCP: Erstellen Sie Wealthy-Context-AI-Apps mit Anthropic” Kurze Kurs von DeepLearning.ai.