面向初学者的FastMCP与Ollama模型上下文协议指南

模型上下文协议，通常简称为MCP，是一种通用语言，用于将大型语言模型与它们自身无法直接访问的工具和数据源连接起来。将其称为“协议”强调了它对对话格式的标准化要求；具体的库则负责实现这一标准，从而避免各团队需要自行设计套接字和消息结构。FastMCP就是下面演示中使用的此类实现之一。

配套仓库地址：https://github.com/harshagangari747/MCPTutorial/tree/main

前置条件

该演示依赖于三个软件包：fastmcp、ollama和langchain-community。推理过程会在本地的qwen3:8b模型上运行，可通过以下命令启动该模型：

ollama run qwen3:8b

准备一个项目文件夹，其中已包含名为 weather_server_mcp.py 和 app.py 的空模板文件，以便服务器和主机有明确的存放位置。

了解 MCP

单独来看，大型语言模型只是一个令牌转换器。令牌进入，令牌离开。除非模型外部有程序执行相应操作，否则它不会调用天气 API、打开数据库或读取系统时间。云服务提供商有时会在其 API 上添加专有的工具运行器，这在实际应用中很方便，但若想了解协议本身则较为不便。通过 Ollama 运行本地模型可以让实验保持独立性。

可以尝试这样的问题：“今天意大利的天气如何？”典型的本地回复会首先承认没有实时的天气数据。即便如此，这句话仍包含了系统需要识别的三个关键信息：主题为天气，“今天”表示日期，而意大利则是地点。模型需要获取或计算天气数据的途径、确定“今天”具体日期的方法，以及将相关天气信息与意大利联系起来的方式。

日历信息的缺失是一个明显的问题。模型无法可靠地判断当前日期。当模型能够提出相应的工具和参数，且运行环境能够实际执行这些工具并返回最新数据，让模型将其整合到答案中时，MCP才发挥作用。

MCP的组成部分

在实际应用中，MCP通常由多个协同工作的组件构成：

1. 可用 API — 任何已能回答相关领域问题的服务，例如网上可找到的天气接口。
2. MCP 服务器 — 一种隐藏 API 或数据库访问方式并发布可调用工具的进程。
3. MCP 主机 — 产品界面，例如结合大语言模型推理与实时数据的智能行程规划器。
4. MCP 客户端 — 存在于主机内部的桥梁，它告知模型有哪些工具可用，将模型的意图转换为 MCP 请求，并将 MCP 的回复转换成模型易于处理的上下文。
5. 传输层 — 使用 JSON-RPC 2.0 协议，当各组件位于远程时通过 HTTP/SSE 传输，而当模型与工具在同一台机器上时则通过 STDIO 传输。
6. 大语言模型 — 此处为 Ollama 提供的 qwen3:8b。

Harness——一种可选的编排工具，能以更少的手动配置将各组件整合在一起；原文以Goose AI作为示例。

明确了这些角色之后，意大利天气查询就变成了一种协同处理过程，而非单一模型调用。

类比

一个生动的比喻有助于明确各部分的职能。驾驶的意图由主机应用体现，大脑则对应大语言模型：它感知道路状况并决定加速、刹车或换挡，但却无法实际踩下踏板。四肢则对应MCP服务器；四肢中的肌肉和骨骼如同独立的工具——一条肢负责转向或换挡，另一条则负责刹车或加速。大脑与肌肉之间的神经连接即为MCP客户端，而传递电信号的神经则相当于传输介质。汽车则对应外部API，整个组装好的躯体则是让各部件协同工作的连接装置。

简化映射关系：

• 大语言模型 → 大脑
• MCP服务器 → 四肢
• 工具 → 肌肉动作
• MCP主机 → 驾驶意图
• MCP客户端 → 神经连接
• 传输介质 → 神经
• 工作API → 汽车
• 连接装置 → 躯体结构

那张图片就足以让服务器、客户端和传输层不会合并成一个模糊的“插件”。

MCP的工作原理

实现时需根据不同角色来构建：服务器、主机、大语言模型、传输层，可选地还包括工具集以及真正的API或服务。服务器负责抽象API并提供各种工具；每个工具都是模型可以请求的单一操作，模型本身从不执行HTTP调用。客户端则负责发布功能列表，并实现双向转换，从而使模型与服务器保持松散耦合。

假设有一台提供get_todays_date()和get_weather_data(city, date)功能的服务器，那么像“巴黎今天的天气如何？”这样的查询可以按以下步骤进行：

1. 模型意识到自己需要今天的日期。
2. 它指示MCP客户端使用get_todays_date函数。
3. 客户端将请求转发给服务器。

服务器负责处理该请求。

客户端为模型重新格式化服务器的回复。

即便获得了日期，模型仍需要巴黎的天气信息。

它会要求客户端传入城市名称和日期后调用get_weather_data函数。

客户端将该请求意图转换为服务器可理解的格式。

服务器调用天气API并返回数据内容。

客户端再次将数据内容转换为模型能理解的格式。

模型最终向用户展示答案。

属于训练数据范围内的历史问题可能仅凭记忆就能回答，但MCP的核心优势在于处理当前上下文——即训练完成后会变化的日期和天气信息。

该项目

该示例让天气相关功能更加具体。MCP服务器负责处理与外部天气API的通信逻辑，而主机应用程序则创建MCP客户端、注册服务器并向Ollama发起查询。将LLM的访问功能隔离在独立的辅助模块中，有助于保持代码结构的清晰可读。

MCP服务器

# MCP Server
# weather_server_mcp.py
from fastmcp import FastMCP
import requests

# This is a server instance that we register in our host
server = FastMCP("weather-mcp-server")


# Third party api data
WEATHER_API_KEY = "api_key_here"
WEATHER_BASE_URL = "https://api.weatherapi.com/v1/"

# Tool 1
@server.tool()
def get_weather_data(city: str) -> float:
    """Get current temperature in Celsius"""
    response = requests.get(
        WEATHER_BASE_URL + "current.json",
        params={"key": WEATHER_API_KEY, "q": city},
    )
    response.raise_for_status()
    return response.json()["current"]["temp_c"]

# Tool 2
@server.tool()
def get_historical_weather_data(city: str, date: str) -> float:
    """Get max temperature for a historical date"""
    response = requests.get(
        WEATHER_BASE_URL + "history.json",
        params={"key": WEATHER_API_KEY, "q": city, "dt": date},
    )
    response.raise_for_status()
    return response.json()["forecast"]["forecastday"][0]["day"]["maxtemp_c"]


if __name__ == "__main__":
    server.run()

那些调用API的函数会被标注上@server.tool()，这样它们就会被作为工具发布出来。每个函数开头的文档字符串并非装饰性内容，而是用于告知模型何时使用该工具。示例中提供了两种工具：一种用于查询某城市的当前天气，另一种用于查询该城市过去某一天的历史天气。

MCP主机、客户端、LLM及传输方式

import asyncio
import sys
import json
from pathlib import Path
from langchain_community.llms import Ollama
from fastmcp import Client
from fastmcp.client.transports import StdioTransport


async def main():
    # We mention the mcp server path.
    server_path = Path(__file__).parent / "weather_server_mcp.py"

    # The transport method here is STDIO
    transport = StdioTransport(
        command=sys.executable,
        args=[str(server_path)],
    )

    # Register the MCP Client
    mcp_client = Client(transport)

    # LLM via Ollama
    llm = Ollama(model="qwen3:8b", temperature=0.5)

    async with mcp_client:
        print("✓ Connected to MCP server!")

        # We can now access that tools are present in the weather server mcp now.
        mcp_tools = await mcp_client.list_tools()
        tools_info = "\n".join([f"- {t.name}: {t.description or t.name}" for t in mcp_tools])

        print(f"✓ Available tools:\n{tools_info}\n")

        # Interactive loop
        while True:
            question = input("🌤️  Ask: ").strip()
            if question.lower() == 'exit':
                break

            try:
                # Step 1: Ask LLM to decide which tool to use
                decision_prompt = f"""Given the question: "{question}"

Available tools:
{tools_info}

Respond with ONLY a JSON object (no other text):
{{"tool": "tool_name", "params": {{"city": "city_name"}}}}

For get_historical_weather_data, use: {{"tool": "get_historical_weather_data", "params": {{"city": "city_name", "date": "YYYY-MM-DD"}}}}"""

                print(f"\n📍 Processing: {question}")
                llm_response = llm.invoke(decision_prompt)

                # Step 2: Parse JSON from LLM response
                json_start = llm_response.find('{')
                json_end = llm_response.rfind('}') + 1

                if json_start == -1 or json_end == 0:
                    print("❌ LLM didn't return valid tool call")
                    continue

                json_str = llm_response[json_start:json_end]
                tool_call = json.loads(json_str)

                print("Tool call: ", tool_call)

                # Handle array responses
                if isinstance(tool_call, list):
                    tool_call = tool_call[0]

                tool_name = tool_call.get("tool")
                params = tool_call.get("params", {})

                print(f"🔧 Calling: {tool_name} with {params}")

                # Step 3: Call MCP tool. This is where we actually call the tool.
                result = await mcp_client.call_tool(tool_name, params)
                answer = result.content[0].text

                print(f"✓ Answer: {answer}°C\n")

            except json.JSONDecodeError as e:
                print(f"❌ JSON parsing error: {e}")
            except Exception as e:
                print(f"❌ Error: {e}\n")


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

具体发生了什么？

首先需要确定主机旁边的服务器模块路径：

server_path = Path(__file__).parent / "weather_server_mcp.py"

构建一个STDIO传输层，使用当前的Python解释器启动该模块：

  # The transport method here is STDIO
    transport = StdioTransport(
        command=sys.executable,
        args=[str(server_path)],
    )

通过该传输层实例化MCP客户端：

mcp_client = Client(transport)

现在，主机已拥有注册的服务器路径、选定的传输层以及客户端。可通过Ollama将模型关联到该系统中：

llm = Ollama(model="qwen3:8b", temperature=0.5)

向客户端请求由weather_server_mcp.py发布的工具目录：

mcp_tools = await mcp_client.list_tools()

将该目录放入提示语中，并指示模型仅回复工具名称和参数。解析完成后，执行选定的工具：

result = await mcp_client.call_tool(tool_name, params)

因此，本教程的核心步骤为：创建服务器、注册服务器、注册客户端、关联大语言模型以及选择传输层。虽然代理框架可以隐藏部分实现细节，但在学习过程中使用简单的循环结构能让每个协议转换步骤都清晰可见。

总体而言，MCP并非单一的库调用，而更像是一种分工协作。模型负责提出建议；客户端负责转换；服务器负责执行操作；传输层负责传递JSON-RPC消息；主机则掌控面向用户的交互流程。一旦这些职责边界明确，要将功能从天气查询替换为日历、CRM系统或内部搜索，基本上只需编写新的工具，并为其提供足够的文档说明，以便模型能够做出正确选择。 在交互流程运行时，要注意观察模型在每次调用工具之前会输出什么。正常的执行轨迹应显示模型使用了真实存在的工具名称，提供了文档中描述的参数键，并在生成面向用户的句子之前等待客户端返回数据。如果模型编造了工具名称，则需要优化提示词或改进工具描述。若服务器出现异常，应通过客户端将错误信息呈现出来，这样模型就可以重新尝试或给出解释，而非产生幻觉内容。

调整天气数据值。这种反馈机制与最初的架构设计同样重要。