Build a Weather‑Query AI Service with FastMCP: Step‑by‑Step Python Guide

1. MCP Server Development

We develop a Server using Python in Stdio mode, requiring uvx installation.

1. Initialize Project

# Initialize project files
uv init muxue-mcp
cd muxue-mcp

# Create virtual environment
uv venv

# Activate environment
.venv\Scripts\activate

# Install dependencies
uv add "mcp[cli]" httpx

2. Write Core Code

We create a weather‑query server named weather_serv.py with a minimal skeleton.

from mcp.server.fastmcp import FastMCP
from pydantic import Field
import httpx, json, os, logging

logger = logging.getLogger("mcp")
mcp = FastMCP("weather")

@mcp.tool(description="Gaode weather query, input city name, return weather")
async def query_weather(city: str = Field(description="city name")) -> str:
    """Gaode weather query"""
    logger.info(f"Received request for city: {city}")
    contents = []
    content = {
        'date': '2025-5-21',
        'week': '4',
        'dayweather': '晴朗',
        'daytemp_float': '14.0',
        'daywind': '北',
        'nightweather': '多云',
        'nighttemp_float': '4.0'
    }
    contents.append(content)
    return json.dumps(contents, ensure_ascii=False)

if __name__ == "__main__":
    mcp.run(transport='stdio')

FastMCP provides tools, resources, prompts, automatic protocol handling and async support.

(2) Initialize FastMCP

# Create a FastMCP instance named "weather"
mcp = FastMCP("weather")

(3) Define Tool

Use @mcp.tool decorator to register a function as a tool that LLM can call.

(4) Main Entry

if __name__ == "__main__":
    mcp.run(transport='stdio')

Running starts the server in stdio mode; SSE mode is introduced later.

2. MCP Client Development

Client code is written in Python and can be integrated into various agents.

1. Initialize Project

# Create project
uv init mcp-client
cd mcp-client
uv venv
.venv\Scripts\activate   # Windows
source .venv/bin/activate   # Unix
uv add mcp python-dotenv openai
del main.py

2. Define MCPClient class

import asyncio, os, json
from typing import Optional, List
from contextlib import AsyncExitStack
from datetime import datetime
import re
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv()
class MCPClient:
    def __init__(self):
        self.exit_stack = AsyncExitStack()
        self.openai_api_key = os.getenv("DASHSCOPE_API_KEY")
        self.base_url = os.getenv("BASE_URL")
        self.model = os.getenv("MODEL")
        if not self.openai_api_key:
            raise ValueError("❌ Missing OpenAI API Key")
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
        self.session: Optional[ClientSession] = None
    async def connect_to_server(self, server_script_path: str):
        # only .py or .js allowed
        is_python = server_script_path.endswith('.py')
        is_js = server_script_path.endswith('.js')
        if not (is_python or is_js):
            raise ValueError("Server script must be .py or .js")
        command = "python" if is_python else "node"
        server_params = StdioServerParameters(command=command, 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()
        tools = (await self.session.list_tools()).tools
        print("
Connected tools:", [t.name for t in tools])
    async def process_query(self, query: str) -> str:
        # prepare tool definitions for the LLM
        tool_resp = await self.session.list_tools()
        available_tools = [{
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": t.inputSchema
            }
        } for t in tool_resp.tools]
        planning_messages = [{"role": "user", "content": query}]
        tool_response = self.client.chat.completions.create(
            model=self.model,
            messages=planning_messages,
            tools=available_tools
        )
        final_text = []
        for choice in tool_response.choices:
            if choice.finish_reason != 'tool_calls':
                final_text.append(choice.message.content)
            else:
                for tool_call in choice.message.tool_calls:
                    tool_name = tool_call.function.name
                    tool_args = tool_call.function.arguments
                    print(f"Calling tool {tool_name} with args {tool_args}")
                    result = await self.session.call_tool(tool_name, json.loads(tool_args))
                    final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
                    if hasattr(choice.message, 'content') and choice.message.content:
                        messages.append({"role": "assistant", "content": choice.message.content})
                    messages.append({"role": "user", "content": result.content})
                    synth_resp = self.client.chat.completions.create(
                        model=self.model,
                        max_tokens=4096,
                        messages=messages
                    )
                    final_text.append(synth_resp.choices[0].message.content)
        return "
".join(final_text)
    async def chat_loop(self):
        print("
🤖 MCP client started! Type 'quit' to exit")
        while True:
            try:
                query = input("
You: ").strip()
                if query.lower() == 'quit':
                    break
                response = await self.process_query(query)
                print(f"
🤖 AI: {response}")
            except Exception as e:
                print(f"
⚠️ Error: {str(e)}")
    async def cleanup(self):
        await self.exit_stack.aclose()

3. Connect to Server

Method connect_to_server builds Stdio parameters, starts the server process and lists available tools.

4. Query Logic

Method process_query sends the user query to the LLM, receives tool calls, executes them via the session, and continues the conversation.

5. Interactive Dialogue

Method chat_loop provides a REPL for user input.

6. Main Entry

async def main():
    server_script_path = r"D:\Test\mcp-project\server.py"
    client = MCPClient()
    try:
        await client.connect_to_server(server_script_path)
        await client.chat_loop()
    finally:
        await client.cleanup()

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

Running python client.py starts the interactive client.

References

https://www.yangyanxing.com/article/use-python-to-develop-mcp-server.html

https://www.yangyanxing.com/article/use-python-to-develop-sse-mcp-server.html

https://modelcontextprotocol.io/quickstart/server

https://modelcontextprotocol.io/quickstart/client