Building AI's Toolkit: A Hands-On Journey with Model Context Protocol (MCP) and Streamlit

AI models are incredible at generating text, images, or even code. But what if we want them to do things in the real world? Like checking the weather, looking up data, or updating a database? This is where the Model Context Protocol (MCP) comes in – it's all about giving AI models a "toolkit" of external abilities.

I recently embarked on building my very first interactive AI tooling application using MCP, and it was a fun, exciting, and incredibly insightful experience. If you're curious about how AI can move from just talking to actually performing actions, read on!

Understanding the Core Idea: AI as a "Doer"

Imagine an AI that needs to find out the current time in New York. It doesn't "know" the time directly. Instead, it needs a "tool" (like a clock API). MCP defines how the AI can discover this "clock tool," understand what information it needs (e.g., "timezone"), send a request, and then receive the result.

My project demonstrates this concept with a simple, end-to-end setup:

1. The AI's Brain (MCP Server): A central hub that knows what "skills" (tools) are available.

2. The AI's Hands (Python Tools): Actual code that performs real-world tasks.

3. The Magic Remote (Streamlit UI): A simple web app that lets YOU choose an AI "skill," provide inputs, and see the real-world result instantly.

Let's dive into how I built this!

Step 1: Project Setup & Dependencies

First, we need a clean environment.

1. Create a Project Directory:

   Bash

   mkdir mcp_tool_explorer cd mcp_tool_explorer

2. Create and Activate a Virtual Environment:

   Bash

   python -m venv .venv # On Windows (Git Bash/CMD): source .venv/Scripts/activate # On Linux/macOS: source .venv/bin/activate

   (Your terminal prompt should now show (.venv).)

3. Install Necessary Libraries:

   • fastapi and uvicorn for building our simple HTTP server.

   • requests for our client to talk to the server.

   • tzdata for timezone information (crucial for Windows).

   • streamlit for our interactive UI.

   Bash

   pip install fastapi uvicorn requests tzdata streamlit

Step 2: Building the MCP Server (The "Toolbox Keeper")

Our server will use FastAPI to define and expose our tools. We'll implement two simple tools: get_current_time and reverse_string.

File: mcp_server_app.py

Python

# mcp_server_app.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import uvicorn
import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError # For time tool

# --- 1. Define Tool Implementations (The 'Hands' that do the work) ---

def get_current_time_impl(timezone: str = "UTC") -> dict:
    """Retrieves the current time for a specified IANA timezone."""
    try:
        now = datetime.datetime.now(ZoneInfo(timezone))
        return {
            "success": True,
            "timezone": timezone,
            "current_time_iso": now.isoformat(),
            "current_time_readable": now.strftime('%Y-%m-%d %H:%M:%S %Z%z')
        }
    except ZoneInfoNotFoundError:
        return {"success": False, "error": f"Invalid timezone: {timezone}"}
    except Exception as e:
        return {"success": False, "error": str(e)}

def reverse_string_impl(input_string: str) -> dict:
    """Reverses a given string."""
    if not isinstance(input_string, str):
        # This part technically won't be hit if Pydantic validation works
        return {"success": False, "error": "Input must be a string."}
    return {
        "success": True,
        "original_string": input_string,
        "reversed_string": input_string[::-1]
    }

# --- 2. Define MCP Tool Metadata (What the AI 'sees' about the tools) ---

# Pydantic models for input validation and schema generation
class GetCurrentTimeParams(BaseModel):
    timezone: str = Field(
        "UTC",
        description="The IANA timezone name (e.g., 'America/New_York', 'Europe/London', 'NZST'). Defaults to UTC."
    )

class ReverseStringParams(BaseModel):
    input_string: str = Field(
        ..., # ... means required
        description="The string to be reversed."
    )

# This dictionary describes our tools for the /tools endpoint
TOOLS_METADATA = {
    "get_current_time": {
        "description": "Retrieves the current time for a specified timezone.",
        "parameters": GetCurrentTimeParams.schema() # Automatically generates JSON Schema
    },
    "reverse_string": {
        "description": "Reverses a given string.",
        "parameters": ReverseStringParams.schema()
    }
}

# --- 3. Initialize FastAPI App (Our conceptual MCP server) ---

app = FastAPI(
    title="Simple MCP Server for Testing",
    description="Exposes tools for time and string manipulation.",
    version="0.1.0"
)

# --- 4. Define API Endpoints for MCP Interaction ---

@app.get("/tools")
async def get_available_tools():
    """
    Endpoint for a model/client to discover available tools and their schemas.
    """
    return {
        "tools": TOOLS_METADATA
    }

class ToolCallRequest(BaseModel):
    tool_name: str
    parameters: dict

@app.post("/call_tool")
async def call_tool(request: ToolCallRequest):
    """
    Endpoint for a model/client to request the execution of a specific tool.
    """
    tool_name = request.tool_name
    params = request.parameters

    if tool_name == "get_current_time":
        try:
            # Validate parameters using Pydantic model
            validated_params = GetCurrentTimeParams(**params)
            result = get_current_time_impl(timezone=validated_params.timezone)
        except Exception as e:
            # Pydantic validation errors often return 422
            raise HTTPException(status_code=422, detail=f"Invalid parameters for get_current_time: {e}")
        return {"tool_name": tool_name, "result": result}

    elif tool_name == "reverse_string":
        try:
            # Validate parameters using Pydantic model
            validated_params = ReverseStringParams(**params)
            result = reverse_string_impl(input_string=validated_params.input_string)
        except Exception as e:
            raise HTTPException(status_code=422, detail=f"Invalid parameters for reverse_string: {e}")
        return {"tool_name": tool_name, "result": result}

    else:
        raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found.")

# --- 5. Run the Server (Entry Point) ---

if __name__ == "__main__":
    print("Starting MCP Server...")
    print("Access interactive API docs at http://127.0.0.1:8000/docs")
    uvicorn.run("mcp_server_app:app", host="127.0.0.1", port=8000, reload=True)

Key Takeaways for the Server:

• FastAPI & Pydantic: These make API creation and automatic input validation (based on defined schemas) incredibly straightforward. This is crucial for robust tool calling.

• /tools endpoint: This is how a smart AI or our UI app discovers what capabilities are available.

• /call_tool endpoint: The central point where requests to execute tools are sent.

• _impl functions: Your actual tool logic, separated from the API definitions.

Step 3: Building the MCP Client & End-to-End Tests

Before building a UI, it's vital to have automated tests to ensure our server and tools work as expected. This client script will simulate an AI calling the tools.

File: mcp_client_test.py

Python

# mcp_client_test.py

import requests
import asyncio # Still needed for asyncio.run in test runner, though client functions are sync
import datetime
import time # For initial delay

# Define the URL of your running MCP server
MCP_SERVER_URL = "http://127.0.0.1:8000"

# --- 1. Basic MCP Client Functions ---

def get_available_tools_from_server(server_url: str = MCP_SERVER_URL) -> dict:
    """Fetches the list of available tools from the MCP server."""
    try:
        response = requests.get(f"{server_url}/tools")
        response.raise_for_status() # Raise an exception for HTTP errors
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching tools: {e}")
        return {"error": str(e)}

def call_tool_on_server(tool_name: str, parameters: dict, server_url: str = MCP_SERVER_URL) -> dict:
    """Calls a specific tool on the MCP server."""
    try:
        payload = {
            "tool_name": tool_name,
            "parameters": parameters
        }
        response = requests.post(f"{server_url}/call_tool", json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error calling tool '{tool_name}': {e}")
        return {"error": str(e)}

# --- 2. Define Your Test Scenarios ---

async def run_test_get_current_time():
    print("\n--- Running Test: get_current_time (Auckland) ---")
    tool_name = "get_current_time"
    params = {"timezone": "Pacific/Auckland"} # Use IANA standard name

    result = call_tool_on_server(tool_name, params) # No await here, function is sync

    if result and "result" in result and result["result"].get("success"):
        print(f"PASS: Time in Auckland - {result['result'].get('current_time_readable')}")
        assert result["result"].get("timezone") == "Pacific/Auckland"
        assert str(datetime.datetime.now().year) in result["result"]["current_time_readable"]
        assert len(result["result"]["current_time_readable"]) > 15
    else:
        print(f"FAIL: Time in Auckland - {result.get('error', 'Unknown error')}")
        assert False, f"Test failed for get_current_time (Auckland): {result}"

async def run_test_get_current_time_invalid_timezone():
    print("\n--- Running Test: get_current_time (Invalid Timezone) ---")
    tool_name = "get_current_time"
    params = {"timezone": "Invalid/Timezone"}

    result = call_tool_on_server(tool_name, params) # No await here

    if result and "result" in result and not result["result"].get("success"):
        print(f"PASS: Handled invalid timezone correctly: {result['result'].get('error')}")
        assert "Invalid timezone" in result["result"]["error"]
    else:
        print(f"FAIL: Did not handle invalid timezone as expected: {result}")
        assert False, f"Test failed for invalid timezone: {result}"

async def run_test_reverse_string_basic():
    print("\n--- Running Test: reverse_string (Basic) ---")
    tool_name = "reverse_string"
    params = {"input_string": "hello"}

    result = call_tool_on_server(tool_name, params) # No await here

    if result and "result" in result and result["result"].get("success"):
        print(f"PASS: Reversed 'hello' to '{result['result'].get('reversed_string')}'")
        assert result["result"]["reversed_string"] == "olleh"
    else:
        print(f"FAIL: Could not reverse string: {result.get('error', 'Unknown error')}")
        assert False, f"Test failed for reverse_string (basic): {result}"

async def run_test_reverse_string_empty():
    print("\n--- Running Test: reverse_string (Empty String) ---")
    tool_name = "reverse_string"
    params = {"input_string": ""}

    result = call_tool_on_server(tool_name, params) # No await here

    if result and "result" in result and result["result"].get("success"):
        print(f"PASS: Reversed empty string correctly.")
        assert result["result"]["reversed_string"] == ""
    else:
        print(f"FAIL: Could not reverse empty string: {result.get('error', 'Unknown error')}")
        assert False, f"Test failed for reverse_string (empty): {result}"

async def run_test_reverse_string_non_string_input():
    print("\n--- Running Test: reverse_string (Non-string Input) ---")
    tool_name = "reverse_string"
    params = {"input_string": 12345} # Invalid input type

    try:
        response = requests.post(f"{MCP_SERVER_URL}/call_tool", json={"tool_name": tool_name, "parameters": params})
        response.raise_for_status()
        print(f"FAIL: Server accepted non-string input unexpectedly. Response: {response.json()}")
        assert False, "Server accepted non-string input unexpectedly"
    except requests.exceptions.HTTPError as e:
        # FastAPI/Pydantic returns 422 (Unprocessable Entity) for validation errors
        if e.response.status_code == 422:
            print(f"PASS: Server rejected non-string input (status 422 - Unprocessable Entity). Detail: {e.response.json().get('detail')}")
            assert "validation error" in str(e.response.json()), "Expected validation error"
        else:
            print(f"FAIL: Unexpected HTTP error for non-string input: {e.response.status_code} - {e.response.text}")
            assert False, f"Unexpected HTTP error: {e}"
    except requests.exceptions.RequestException as e:
        print(f"FAIL: Request error for non-string input: {e}")
        assert False, f"Request error: {e}"

# --- 3. Main Test Runner ---

async def run_all_e2e_tests():
    print("--- Starting End-to-End MCP Tests ---")

    print("\n--- Discovering Tools ---")
    tools_info = get_available_tools_from_server() # No await here
    if tools_info and "tools" in tools_info:
        print("Successfully discovered tools:")
        for name, data in tools_info["tools"].items():
            print(f"  - {name}: {data['description']}")
    else:
        print("Failed to discover tools. Ensure server is running.")
        return

    # Run individual test functions
    await run_test_get_current_time()
    await run_test_get_current_time_invalid_timezone()
    await run_test_reverse_string_basic()
    await run_test_reverse_string_empty()
    await run_test_reverse_string_non_string_input()

    print("\n--- All End-to-End MCP Tests Completed ---")

if __name__ == "__main__":
    print("WARNING: Ensure your MCP server (mcp_server_app.py) is running in a separate terminal!")
    print(f"  Expected server URL: {MCP_SERVER_URL}")
    print("\nStarting tests in 3 seconds...")
    time.sleep(3)
    asyncio.run(run_all_e2e_tests())

Testing Tip: This script is designed to run in a separate terminal while your mcp_server_app.py is active. Those glorious "PASS" messages were a sweet reward!

Step 4: Building the Streamlit UI (The "Magic Remote")

Now for the fun part: making an interactive frontend where you can play with your AI tools!

File: mcp_ui_app.py

Python

# mcp_ui_app.py

import streamlit as st
import requests
import json # For pretty printing JSON

# Define the URL of your running MCP server
MCP_SERVER_URL = "http://127.0.0.1:8000"

# --- MCP Client Functions (Synchronous for Streamlit compatibility) ---

def get_available_tools_from_server(server_url: str = MCP_SERVER_URL) -> dict:
    """Fetches the list of available tools from the MCP server."""
    try:
        response = requests.get(f"{server_url}/tools")
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        st.error(f"Error fetching tools from server at {server_url}: {e}")
        return {"error": str(e)}

def call_tool_on_server(tool_name: str, parameters: dict, server_url: str = MCP_SERVER_URL) -> dict:
    """Calls a specific tool on the MCP server."""
    try:
        payload = {
            "tool_name": tool_name,
            "parameters": parameters
        }
        response = requests.post(f"{server_url}/call_tool", json=payload)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        st.error(f"HTTP Error calling tool '{tool_name}': Status {e.response.status_code}")
        try:
            error_details = e.response.json()
            st.json(error_details)
            return {"error": error_details}
        except json.JSONDecodeError:
            st.error(f"Server responded with: {e.response.text}")
            return {"error": e.response.text}
    except requests.exceptions.RequestException as e:
        st.error(f"Network/Request Error calling tool '{tool_name}': {e}")
        return {"error": str(e)}


# --- Streamlit UI Components ---

st.set_page_config(page_title="MCP Tool Explorer", layout="wide")
st.title("🤖 Model Context Protocol (MCP) Tool Explorer")
st.markdown("Interact with your deployed MCP server and its tools.")

st.sidebar.header("Server Settings")
server_url_input = st.sidebar.text_input("MCP Server URL", MCP_SERVER_URL)
if server_url_input != MCP_SERVER_URL:
    MCP_SERVER_URL = server_url_input
    st.sidebar.success(f"Server URL set to: {MCP_SERVER_URL}")


# --- Tool Discovery ---
st.header("Available Tools")
@st.cache_data(ttl=300) # Cache the tool list for 5 minutes for performance
def get_tools_cached(url):
    return get_available_tools_from_server(url)

tools_data = get_tools_cached(MCP_SERVER_URL)

if not tools_data or "error" in tools_data:
    st.error("Could not connect to the MCP server or fetch tools. Please ensure the server is running.")
    st.info(f"Attempting to connect to: {MCP_SERVER_URL}")
    st.stop()

available_tools = tools_data.get("tools", {})
if not available_tools:
    st.warning("No tools found on the server.")
    st.stop()

tool_names = list(available_tools.keys())
selected_tool_name = st.selectbox("Select a Tool:", tool_names)

# --- Tool Interaction Section ---
if selected_tool_name:
    tool_metadata = available_tools[selected_tool_name]
    st.subheader(f"Call Tool: `{selected_tool_name}`")
    st.markdown(f"**Description:** {tool_metadata.get('description', 'No description provided.')}")

    params_schema = tool_metadata.get('parameters', {}).get('properties', {})
    required_params = tool_metadata.get('parameters', {}).get('required', [])

    tool_parameters = {}
    st.write("Parameters:")
    if params_schema:
        for param_name, param_details in params_schema.items():
            param_type = param_details.get('type', 'string')
            param_description = param_details.get('description', '')
            param_default = param_details.get('default')
            is_required = param_name in required_params

            key = f"{selected_tool_name}_{param_name}" # Unique key for Streamlit widgets

            with st.expander(f"**{param_name}** {'(Required)' if is_required else ''} - *{param_type}*"):
                st.write(param_description)
                # Streamlit widgets based on parameter type
                if param_type == "string":
                    tool_parameters[param_name] = st.text_input(
                        f"Enter {param_name}",
                        value=param_default if param_default is not None else "",
                        key=key
                    )
                elif param_type == "integer":
                    tool_parameters[param_name] = st.number_input(
                        f"Enter {param_name}",
                        value=param_default if param_default is not None else 0,
                        step=1,
                        key=key
                    )
                elif param_type == "number": # For floats
                    tool_parameters[param_name] = st.number_input(
                        f"Enter {param_name}",
                        value=param_default if param_default is not None else 0.0,
                        key=key
                    )
                elif param_type == "boolean":
                    tool_parameters[param_name] = st.checkbox(
                        f"{param_name}?",
                        value=param_default if param_default is not None else False,
                        key=key
                    )
                else:
                    st.warning(f"Unsupported parameter type '{param_type}' for '{param_name}'. Using text input.")
                    tool_parameters[param_name] = st.text_input(
                        f"Enter {param_name} (unsupported type)",
                        value=str(param_default) if param_default is not None else "",
                        key=key
                    )
    else:
        st.info("This tool requires no parameters.")


    if st.button(f"Execute {selected_tool_name}", type="primary"):
        st.subheader("Tool Execution Result:")
        # Filter out empty string parameters if not required by schema
        # Streamlit widgets return appropriate types (int, float, bool)
        # So explicit type conversion from string isn't strictly needed if using number_input/checkbox
        final_params = {k: v for k, v in tool_parameters.items() if v != "" or k in required_params}

        with st.spinner(f"Calling {selected_tool_name} on server..."):
            tool_result = call_tool_on_server(selected_tool_name, final_params, MCP_SERVER_URL)

            if tool_result and "error" not in tool_result:
                st.success("Tool executed successfully!")
                st.json(tool_result) # Display the raw JSON result
                if tool_result.get("result", {}).get("success"):
                    st.write("### Structured Result:")
                    st.json(tool_result["result"])
                else:
                    st.error("Tool execution failed (server reported failure):")
                    st.json(tool_result["result"])
            else:
                st.error("Failed to execute tool. Check error messages above.")

UI Magic: Streamlit makes it incredibly easy to turn Python code into a web app. The dynamic parameter generation, error handling in the UI, and caching for performance are key features.

How to Run Everything!

1. Start Your MCP Server:

   • Open a first terminal in your mcp_tool_explorer directory (with .venv activated).

   • Run: python mcp_server_app.py

   • Keep this terminal running.

2. Run Your End-to-End Tests (Optional, but Recommended!):

   • Open a second terminal in your mcp_tool_explorer directory (with .venv activated).

   • Run: python mcp_client_test.py

   • Watch those green "PASS" messages fly by!

3. Launch Your Streamlit UI:

   • Open a third terminal in your mcp_tool_explorer directory (with .venv activated).

   • Run: streamlit run mcp_ui_app.py

   • A new browser tab will open (usually http://localhost:8501).

Now, interact with your MCP tools directly from the web interface!

Key Learnings from This Project:

• MCP as a Bridge: Understanding how a protocol like MCP connects an AI's intent with executable code is fundamental for future AI agents.

• API Design Matters: Clear tool definitions and robust API endpoints (thanks, FastAPI!) are crucial for seamless integration.

• Validation is Gold: Pydantic's automatic input validation was a lifesaver, ensuring our tools received the correct data types.

• UIs for Debugging & Demo: Streamlit proved invaluable not just for a pretty demo, but for easily testing various inputs and visualizing the structured outputs from my tools. It makes complex interactions so much more tangible.

• E2E Testing is Non-Negotiable: Automated tests catch issues early and provide confidence that your entire pipeline works.

This project was a fantastic stepping stone into the exciting world of actionable AI. It highlighted the power of modular design and the joy of seeing complex systems come to life!

Happy coding!