本文档以实现一个贷款利息计算器为示例说明mcp的用法。

server 代码  

#!/usr/bin/env python3
"""MCP Server implementation for Loan Calculator."""

import asyncio
import math
from typing import Any

from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
)

# 创建MCP Server实例
server = Server("loan-calculator-server")


def calculate_loan_equal_principal_interest(principal: float, years: int, annual_rate: float) -> dict[str, Any]:
    """
    计算等额本息贷款的月供、总利息和本息合计。

    等额本息：每月还款额固定，前期利息多、本金少，后期逐渐反转。

    Args:
        principal: 贷款本金（元）
        years: 贷款期限（年）
        annual_rate: 年利率（百分比，如 4.5 表示 4.5%）

    Returns:
        包含计算结果的字典
    """
    # 将年利率转换为月利率（小数形式）
    monthly_rate = annual_rate / 100 / 12

    # 贷款总月数
    months = years * 12

    # 计算月供（等额本息公式）
    # 月供 = [本金 × 月利率 × (1 + 月利率)^还款月数] / [(1 + 月利率)^还款月数 - 1]
    if monthly_rate == 0:
        monthly_payment = principal / months
    else:
        monthly_payment = (principal * monthly_rate * (1 + monthly_rate) ** months) / \
                         ((1 + monthly_rate) ** months - 1)

    # 计算总还款额
    total_payment = monthly_payment * months

    # 计算总利息
    total_interest = total_payment - principal

    return {
        "monthly_payment": round(monthly_payment, 2),
        "monthly_payment_first": round(monthly_payment, 2),
        "monthly_payment_last": round(monthly_payment, 2),
        "monthly_payment_decline": 0,
        "total_interest": round(total_interest, 2),
        "total_payment": round(total_payment, 2),
        "principal": principal,
        "years": years,
        "annual_rate": annual_rate,
        "method": "等额本息"
    }


def calculate_loan_equal_principal(principal: float, years: int, annual_rate: float) -> dict[str, Any]:
    """
    计算等额本金贷款的月供、总利息和本息合计。

    等额本金：每月还本金固定，利息逐月递减，首月还款最高。

    Args:
        principal: 贷款本金（元）
        years: 贷款期限（年）
        annual_rate: 年利率（百分比，如 4.5 表示 4.5%）

    Returns:
        包含计算结果的字典
    """
    # 将年利率转换为月利率（小数形式）
    monthly_rate = annual_rate / 100 / 12

    # 贷款总月数
    months = years * 12

    # 每月偿还本金
    monthly_principal = principal / months

    # 首月利息 = 全部本金 × 月利率
    first_month_interest = principal * monthly_rate

    # 末月利息 = 剩余本金 × 月利率 = (每月本金) × 月利率
    last_month_interest = monthly_principal * monthly_rate

    # 首月还款 = 每月本金 + 首月利息
    first_month_payment = monthly_principal + first_month_interest

    # 末月还款 = 每月本金 + 末月利息
    last_month_payment = monthly_principal + last_month_interest

    # 每月还款递减额 = 每月本金 × 月利率
    monthly_decline = monthly_principal * monthly_rate

    # 总利息 = (首月利息 + 末月利息) × 月数 / 2
    total_interest = (first_month_interest + last_month_interest) * months / 2

    # 总还款额 = 本金 + 总利息
    total_payment = principal + total_interest

    return {
        "monthly_payment": round((first_month_payment + last_month_payment) / 2, 2),
        "monthly_payment_first": round(first_month_payment, 2),
        "monthly_payment_last": round(last_month_payment, 2),
        "monthly_payment_decline": round(monthly_decline, 2),
        "total_interest": round(total_interest, 2),
        "total_payment": round(total_payment, 2),
        "principal": principal,
        "years": years,
        "annual_rate": annual_rate,
        "method": "等额本金"
    }


def calculate_loan(principal: float, years: int, annual_rate: float,
                   repayment_method: str = "等额本息") -> dict[str, Any]:
    """
    计算贷款月供、总利息和本息合计。

    Args:
        principal: 贷款本金（元）
        years: 贷款期限（年）
        annual_rate: 年利率（百分比，如 4.5 表示 4.5%）
        repayment_method: 还款方式，"等额本息" 或 "等额本金"

    Returns:
        包含计算结果的字典
    """
    if repayment_method == "等额本金":
        return calculate_loan_equal_principal(principal, years, annual_rate)
    else:
        return calculate_loan_equal_principal_interest(principal, years, annual_rate)


def format_result(result: dict[str, Any]) -> str:
    """格式化计算结果为易读的字符串。"""
    method = result['method']

    if method == "等额本金":
        return f"""贷款计算结果（{method}）

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
贷款本金：¥{result['principal']:,.2f} 元
贷款期限：{result['years']} 年 ({result['years'] * 12} 个月)
年利率：{result['annual_rate']}%
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
首月还款：¥{result['monthly_payment_first']:,.2f} 元
末月还款：¥{result['monthly_payment_last']:,.2f} 元
每月递减：¥{result['monthly_payment_decline']:,.2f} 元
总利息：¥{result['total_interest']:,.2f} 元
本息合计：¥{result['total_payment']:,.2f} 元
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"""
    else:
        return f"""贷款计算结果（{method}）

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
贷款本金：¥{result['principal']:,.2f} 元
贷款期限：{result['years']} 年 ({result['years'] * 12} 个月)
年利率：{result['annual_rate']}%
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
每月还款：¥{result['monthly_payment']:,.2f} 元（固定）
总利息：¥{result['total_interest']:,.2f} 元
本息合计：¥{result['total_payment']:,.2f} 元
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"""


@server.list_tools()
async def list_tools() -> list[Tool]:
    """列出可用的工具。"""
    return [
        Tool(
            name="calculate_loan",
            description="计算贷款的月供、总利息和本息合计。支持等额本息（每月还款固定）和等额本金（首月最高，逐月递减）两种还款方式。",
            inputSchema={
                "type": "object",
                "properties": {
                    "principal": {
                        "type": "number",
                        "description": "贷款本金，单位：元（例如：100000 表示10万元）"
                    },
                    "years": {
                        "type": "integer",
                        "description": "贷款期限，单位：年（例如：30 表示30年）"
                    },
                    "annual_rate": {
                        "type": "number",
                        "description": "年利率，单位：%（例如：4.5 表示4.5%）"
                    },
                    "repayment_method": {
                        "type": "string",
                        "description": "还款方式：等额本息（每月还款固定）或等额本金（首月最高，逐月递减），默认为等额本息",
                        "enum": ["等额本息", "等额本金"],
                        "default": "等额本息"
                    }
                },
                "required": ["principal", "years", "annual_rate"]
            }
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """处理工具调用。"""
    if name == "calculate_loan":
        try:
            principal = float(arguments["principal"])
            years = int(arguments["years"])
            annual_rate = float(arguments["annual_rate"])
            repayment_method = arguments.get("repayment_method", "等额本息")

            # 参数验证
            if principal <= 0:
                return [TextContent(
                    type="text",
                    text="错误：贷款本金必须大于0"
                )]
            if years <= 0:
                return [TextContent(
                    type="text",
                    text="错误：贷款期限必须大于0"
                )]
            if annual_rate < 0:
                return [TextContent(
                    type="text",
                    text="错误：年利率不能为负数"
                )]
            if repayment_method not in ["等额本息", "等额本金"]:
                return [TextContent(
                    type="text",
                    text="错误：还款方式必须是'等额本息'或'等额本金'"
                )]

            result = calculate_loan(principal, years, annual_rate, repayment_method)
            return [TextContent(
                type="text",
                text=format_result(result)
            )]

        except KeyError as e:
            return [TextContent(
                type="text",
                text=f"错误：缺少必需参数 {e}"
            )]
        except (ValueError, TypeError) as e:
            return [TextContent(
                type="text",
                text=f"错误：参数类型不正确 - {e}"
            )]
        except Exception as e:
            return [TextContent(
                type="text",
                text=f"计算错误：{e}"
            )]

    return [TextContent(
        type="text",
        text=f"未知工具: {name}"
    )]


async def main():
    """启动MCP Server。"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="loan-calculator-server",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )


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

client-1.py 代码  （直接调用）

"""MCP Client for Loan Calculator - 直接调用MCP工具"""

import asyncio
import os
from dotenv import load_dotenv

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

# 加载环境变量
load_dotenv()


async def main():
    """直接调用MCP Server的贷款计算工具"""

    # 连接到MCP Server（通过stdio）
    server_params = StdioServerParameters(
        command="uv",
        args=[
            "run",
            "python",
            "-m",
            "mcp_server_client.server"
        ],
        env=os.environ.copy()
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化
            await session.initialize()

            # 列出可用工具
            tools = await session.list_tools()
            print("=" * 50)
            print("可用的工具:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            print("=" * 50)
            print()

            # 示例1: 计算100万贷款，30年，4.5%利率
            print("示例1: 贷款100万，30年，年利率4.5%")
            print("-" * 50)
            result = await session.call_tool(
                "calculate_loan",
                arguments={
                    "principal": 1000000,
                    "years": 30,
                    "annual_rate": 4.5
                }
            )
            for content in result.content:
                print(content.text)
            print()

            # 示例2: 计算50万贷款，20年，3.8%利率
            print("示例2: 贷款50万，20年，年利率3.8%")
            print("-" * 50)
            result = await session.call_tool(
                "calculate_loan",
                arguments={
                    "principal": 500000,
                    "years": 20,
                    "annual_rate": 3.8
                }
            )
            for content in result.content:
                print(content.text)
            print()

            # 示例3: 计算200万贷款，25年，4.2%利率
            print("示例3: 贷款200万，25年，年利率4.2%")
            print("-" * 50)
            result = await session.call_tool(
                "calculate_loan",
                arguments={
                    "principal": 2000000,
                    "years": 25,
                    "annual_rate": 4.2
                }
            )
            for content in result.content:
                print(content.text)


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

client-2 代码 （LLM调用mcp工具）

#!/usr/bin/env python3
"""MCP Client for Loan Calculator - 结合DeepSeek大模型调用"""

import asyncio
import json
import os
from dotenv import load_dotenv

from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from openai import AsyncOpenAI

# 加载环境变量
load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = "https://api.deepseek.com"


async def main():
    """使用DeepSeek大模型结合MCP Server进行智能贷款计算咨询"""

    # 初始化DeepSeek客户端
    client = AsyncOpenAI(
        api_key=DEEPSEEK_API_KEY,
        base_url=DEEPSEEK_BASE_URL
    )

    # 连接到MCP Server
    server_params = StdioServerParameters(
        command="uv",
        args=[
            "run",
            "python",
            "-m",
            "mcp_server_client.server"
        ],
        env=os.environ.copy()
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化MCP会话
            await session.initialize()

            # 获取可用工具
            tools = await session.list_tools()
            print("=" * 50)
            print("MCP Server 工具列表:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            print("=" * 50)
            print()

            # 将MCP工具转换为OpenAI函数格式
            mcp_tools = [
                {
                    "type": "function",
                    "function": {
                        "name": "calculate_loan",
                        "description": "计算贷款的月供、总利息和本息合计（等额本息还款方式）",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "principal": {
                                    "type": "number",
                                    "description": "贷款本金，单位：元（例如：100000 表示10万元）"
                                },
                                "years": {
                                    "type": "integer",
                                    "description": "贷款期限，单位：年（例如：30 表示30年）"
                                },
                                "annual_rate": {
                                    "type": "number",
                                    "description": "年利率，单位：%（例如：4.5 表示4.5%）"
                                }
                            },
                            "required": ["principal", "years", "annual_rate"]
                        }
                    }
                }
            ]

            # 示例对话
            questions = [
                "我想贷款100万元买房，贷款30年，年利率4.5%，请帮我计算每月需要还款多少钱？总共利息是多少？",
                "如果我有50万预算，贷款20年，年利率3.8%，月供多少？和30年贷款相比有什么区别？",
            ]

            for i, question in enumerate(questions, 1):
                print(f"问题 {i}: {question}")
                print("-" * 50)

                # 调用DeepSeek
                response = await client.chat.completions.create(
                    model="deepseek-chat",
                    messages=[
                        {"role": "system", "content": "你是一个专业的贷款计算助手。当用户询问贷款相关问题时，使用calculate_loan工具来计算准确的数字。计算完成后，用友好的方式解释结果给用户。"},
                        {"role": "user", "content": question}
                    ],
                    tools=mcp_tools,
                    tool_choice="auto"
                )

                # 处理响应
                message = response.choices[0].message

                # 如果需要调用工具
                if message.tool_calls:
                    for tool_call in message.tool_calls:
                        if tool_call.function.name == "calculate_loan":
                            # 解析参数
                            function_args = json.loads(tool_call.function.arguments)

                            # 调用MCP工具
                            result = await session.call_tool(
                                "calculate_loan",
                                arguments=function_args
                            )

                            # 获取工具结果
                            tool_result = result.content[0].text

                            print(tool_result)
                            print()

                            # 让DeepSeek解释结果
                            explanation_response = await client.chat.completions.create(
                                model="deepseek-chat",
                                messages=[
                                    {"role": "system", "content": "你是一个专业的贷款计算助手。请用友好的方式解释贷款计算结果。"},
                                    {"role": "user", "content": question},
                                    {"role": "assistant", "content": None, "tool_calls": [tool_call]},
                                    {"role": "tool", "content": tool_result, "tool_call_id": tool_call.id}
                                ]
                            )

                            print("AI 解释:")
                            print(explanation_response.choices[0].message.content)
                else:
                    print(message.content)

                print("\n" + "=" * 50 + "\n")

            # 交互式问答
            print("进入交互模式（输入 'quit' 退出）:")
            print("=" * 50)

            while True:
                user_input = input("\n请输入您的问题: ").strip()

                if user_input.lower() in ['quit', 'exit', '退出']:
                    print("再见！")
                    break

                if not user_input:
                    continue

                response = await client.chat.completions.create(
                    model="deepseek-chat",
                    messages=[
                        {"role": "system", "content": "你是一个专业的贷款计算助手。当用户询问贷款相关问题时，使用calculate_loan工具来计算准确的数字。计算完成后，用友好的方式解释结果给用户。"},
                        {"role": "user", "content": user_input}
                    ],
                    tools=mcp_tools,
                    tool_choice="auto"
                )

                message = response.choices[0].message

                if message.tool_calls:
                    for tool_call in message.tool_calls:
                        if tool_call.function.name == "calculate_loan":
                            function_args = json.loads(tool_call.function.arguments)
                            result = await session.call_tool(
                                "calculate_loan",
                                arguments=function_args
                            )
                            tool_result = result.content[0].text

                            print("\n计算结果:")
                            print(tool_result)
                            print()

                            explanation_response = await client.chat.completions.create(
                                model="deepseek-chat",
                                messages=[
                                    {"role": "system", "content": "你是一个专业的贷款计算助手。请用友好的方式解释贷款计算结果。"},
                                    {"role": "user", "content": user_input},
                                    {"role": "assistant", "content": None, "tool_calls": [tool_call]},
                                    {"role": "tool", "content": tool_result, "tool_call_id": tool_call.id}
                                ]
                            )

                            print("AI 解释:")
                            print(explanation_response.choices[0].message.content)
                else:
                    print(f"\n回答: {message.content}")


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