Complete MCP Tutorial: How to Build a Local MCP Voice Assistant in Python

# Create project directory
mkdir mcp-voice-assistant
cd mcp-voice-assistant

# Create and activate virtual environment
# macOS/Linux:
python3 -m venv .venv
source .venv/bin/activate

# Windows:
python -m venv .venv
.venv\Scripts\activate

# Install all dependencies
pip install mcp python-dotenv requests picollm pvcheetah pvorca pvrecorder pvspeaker

# Create project files
# macOS/Linux:
touch .env server.py client.py

# Windows:
New-Item .env -ItemType File
New-Item server.py -ItemType File
New-Item client.py -ItemType File

WEATHER_API_KEY="${YOUR_WEATHER_API_KEY}"

import os
import requests
import json

from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# Load environment variables
load_dotenv()
api_key = os.getenv("WEATHER_API_KEY")
base_url = "https://api.weatherapi.com/v1"

# Initialize FastMCP server
server = FastMCP("Weather Voice Assistant")

def is_use_fahrenheit(country: str) -> bool:
    """Determine if country uses Fahrenheit for temperature"""
    fahrenheit_countries = [
        "United States of America",
        "Bahamas",
        "Cayman Islands",
        "Palau",
        "Micronesia",
        # ...
    ]
    return country in fahrenheit_countries

@server.tool()
def fetch_weather(city_name: str) -> dict:
    """
    Retrieves the current weather conditions for a given city.

    Args:
        city_name: Name of the city (e.g., "Vancouver", "San Francisco")

    Returns:
        A dictionary containing the current weather conditions data.
    """
    try:
        response = requests.get(f"{base_url}/current.json?key={api_key}&q={city_name}")
        weather_data = response.json()

        # Extract and format current weather data
        location = weather_data["location"]["name"]
        region = weather_data["location"]["region"]
        country = weather_data["location"]["country"]
        temp_f = str(weather_data["current"]["temp_f"]) + "°F"
        temp_c = str(weather_data["current"]["temp_c"]) + "°C"
        temp = temp_f if is_use_fahrenheit(country) else temp_c
        condition = weather_data["current"]["condition"]["text"]

        result = f"Weather in {location}, {region}, {country}: Temperature: {temp}, Condition: {condition}"

        if response.status_code == 200:
            return {"success": True, "contents": result}
        else:
            return {
                "success": False,
                "error": f"API request failed with status code: {response.status_code}",
            }
    except Exception as e:
        return {"success": False, "error": str(e)}

@server.tool()
def fetch_forecast(city_name: str, days: int) -> dict:
    """
    Retrieves the weather forecast for a given city over a specified number of days.

    Args:
        city_name: Name of the city (e.g., "Vancouver", "San Francisco")
        days: Number of days to retrieve the forecast for (maximum 14).

    Returns:
        A dictionary containing the weather forecast data.
    """
    try:
        response = requests.get(
            f"{base_url}/forecast.json?key={api_key}&q={city_name}&days={days}"
        )
        weather_data = response.json()

        # Extract and format weather forecast data
        location = weather_data["location"]["name"]
        region = weather_data["location"]["region"]
        country = weather_data["location"]["country"]

        days = [
            {
                "date": day["date"],
                "avgtemp": (
                    f"{day['day']['avgtemp_f']}°F"
                    if is_use_fahrenheit(country)
                    else f"{day['day']['avgtemp_c']}°C"
                ),
                "condition": day["day"]["condition"]["text"],
            }
            for day in weather_data["forecast"]["forecastday"]
        ]

        result = f"Forecast for {location}, {region}, {country}: " + " ".join(
            f"On {day['date']}, the average temperature is {day['avgtemp']} and the condition is {day['condition']}."
            for day in days
        )

        if response.status_code == 200:
            return {"success": True, "contents": result}
        else:
            return {
                "success": False,
                "error": f"API request failed with status code: {response.status_code}",
            }
    except Exception as e:
        return {"success": False, "error": str(e)}

def main():
    server.run(transport="stdio")


if __name__ == "__main__":
    main()

PICOVOICE_ACCESS_KEY="${YOUR_PICOVOICE_ACCESS_KEY}"

import argparse
import asyncio
import ast
import os
import re
from collections import deque
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from dotenv import load_dotenv

import picollm
import pvcheetah
import pvorca
from pvrecorder import PvRecorder
from pvspeaker import PvSpeaker

# Load environment variables
load_dotenv()
access_key = os.getenv("PICOVOICE_ACCESS_KEY")

class MCPClient:
    def __init__(self, pllm_model_path):
        # MCP session and async cleanup stack
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        
        # Local LLM
        self.pllm = picollm.create(access_key=access_key, model_path=pllm_model_path)
        
        # Initialize streaming STT engine
        self.cheetah = pvcheetah.create(
            access_key=access_key,
            endpoint_duration_sec=2.0,
            enable_automatic_punctuation=True,
        )
        
        # Recorder for capturing microphone input
        self.recorder = PvRecorder(frame_length=self.cheetah.frame_length)
        
        # Initialize streaming TTS engine
        self.orca = pvorca.create(access_key=access_key)
        
        # Speaker for audio playback
        self.speaker = PvSpeaker(
            sample_rate=self.orca.sample_rate,
            bits_per_sample=16,
        )

async def connect_to_server(self, server_script_path: str):
    """Connect to an MCP server"""
    server_params = StdioServerParameters(
        command="python3", args=[server_script_path], env=None
    )

    stdio_transport = await self.exit_stack.enter_async_context(
        stdio_client(server_params)
    )
    self.stdio, self.write = stdio_transport
    self.session = await self.exit_stack.enter_async_context(
        ClientSession(self.stdio, self.write)
    )

    await self.session.initialize()

    # List available tools
    response = await self.session.list_tools()
    tools = response.tools
    print("\nConnected to server with tools:", [tool.name for tool in tools])

# Add this helper function before the MCPClient class
def parse_args(args_str: str) -> dict:
    """Parse function arguments from string format"""
    try:
        # Handle simple key=value format
        args_dict = {}
        pairs = args_str.split(',')
        for pair in pairs:
            key, value = pair.split('=')
            key = key.strip()
            value = value.strip().strip('"').strip("'")
            # Try to convert to appropriate type
            try:
                value = ast.literal_eval(value)
            except (ValueError, SyntaxError):
                pass
            args_dict[key] = value
        return args_dict
    except Exception as e:
        print(f"Error parsing args: {e}")
        return {}

# Add this regex pattern at module level
pattern = r"""
    (?P<func>[a-zA-Z_][a-zA-Z0-9_]*)   # function name
    \s*\(\s*                           # opening parenthesis
    (?P<args>[^)]*)                    # arguments
    \s*\)                              # closing parenthesis
"""

async def process_query(self, query: str) -> str:
    """Process a query using picoLLM and available tools"""

    response = await self.session.list_tools()
    available_tools = [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema,
        }
        for tool in response.tools
    ]

    prompt = """
<|begin_of_text|><|start_header_id|>user<|end_header_id|>
Questions: {query}
Here is a list of functions in JSON format that you can invoke:
{available_tools}
Should you decide to return the function call(s), Put it in the format of: [func1(params_name=params_value, params_name2=params_value2...), func2(params)]
NO other text MUST be included.<|eot_id|><|start_header_id|>assistant<|end_header_id|>
    """.format(
        query=query, available_tools=available_tools
    )

    dialog = self.pllm.get_dialog(history=1)
    dialog.add_human_request(prompt)

    # Initial picoLLM API call
    response = self.pllm.generate(
        prompt=dialog.prompt(),
        completion_token_limit=1000,
        presence_penalty=0.0,
        frequency_penalty=0.0,
        temperature=0.0,
        top_p=1.0,
    )

    match = re.search(pattern, response.completion, re.VERBOSE)

    if match:
        func_name = match.group("func")
        args = match.group("args")

        # Parse arguments and call tool
        args_dict = parse_args(args)
        result = await self.session.call_tool(func_name, args_dict)
        dialog.add_llm_response(response.completion)

        execution_result = result.content

        followup_prompt = f"""
<|begin_of_text|><|start_header_id|>user<|end_header_id|>
The user requested weather information, and your app executed it.
Here is the result of that execution:
{execution_result}

Please summarize this result for the user in a conversational, easy-to-understand message.
Only return the text, no function calls.
<|eot_id|><|start_header_id|>assistant<|end_header_id|>
"""
        dialog.add_human_request(followup_prompt)

        response = self.pllm.generate(
            prompt=dialog.prompt(),
            completion_token_limit=1000,
            presence_penalty=0.0,
            frequency_penalty=0.0,
            temperature=0.0,
            top_p=1.0,
        )
        return response.completion.removesuffix("<|eot_id|>")

    else:
        return f"No tools called. LLM response: {response.completion}"

async def chat_loop(self):
    """Interactive voice chat loop"""
    print("\nMCP Voice Assistant Started!")
    print("Speak your queries or type 'stop' to exit.")

    while True:
        try:
            # Ask user for text input first
            action = input("\n> Press 'Enter' to speak, type 'stop' to exit: ").strip()
            if action.lower() == "stop":
                print("Exiting...")
                break

            print("\nAsk MCP Assistant: ", end="")
            query = ""

            # Enable the user to provide their query by speaking
            self.recorder.start()
            while True:
                frame = self.recorder.read()
                partial_transcript, is_endpoint = self.cheetah.process(frame)
                query += partial_transcript
                print(partial_transcript, end="", flush=True)

                if is_endpoint:
                    final_transcript = self.cheetah.flush()
                    query += final_transcript
                    print(final_transcript + "\n")
                    break
            self.recorder.stop()

            # Process query through MCP + LLM
            response = await self.process_query(query)
            print(f"\nMCP Assistant: {response}")

            # Speak the response back to the user
            self.speaker.start()
            response_pcm, _ = self.orca.synthesize(response)

            if response_pcm is not None:
                pcm_buffer = deque()
                pcm_buffer.append(response_pcm)

                while len(pcm_buffer) > 0:
                    pcm = pcm_buffer.popleft()
                    written = self.speaker.write(pcm)
                    if written < len(pcm):
                        pcm_buffer.appendleft(pcm[written:])
                self.speaker.flush()

            self.speaker.stop()

        except KeyboardInterrupt:
            print("\nInterrupted by user")
            break
        except Exception as e:
            print(f"\nError: {str(e)}")

async def cleanup(self):
    """Release all resources for STT, TTS, and LLM"""
    await self.exit_stack.aclose()
    self.recorder.delete()   # Release microphone
    self.cheetah.delete()    # Release STT engine
    self.speaker.delete()    # Release speaker
    self.orca.delete()       # Release TTS engine
    self.pllm.release()      # Release LLM

async def main():
    parser = argparse.ArgumentParser(description="MCP Weather Voice Assistant Client")
    parser.add_argument(
        "-s",
        "--server_script",
        help="Path to the MCP server script",
        required=True,
    )
    parser.add_argument(
        "-m",
        "--pllm_model_path",
        help="Absolute path to the file containing LLM parameters.",
        required=True,
    )

    args = parser.parse_args()

    client = MCPClient(pllm_model_path=args.pllm_model_path)
    try:
        await client.connect_to_server(args.server_script)
        await client.chat_loop()
    finally:
        await client.cleanup()


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

python client.py -s server.py -m ./models/llama-3.2-3b-instruct-505.pllm