from fastmcp import FastMCP

# 创建服务器实例，名字会显示给客户端
mcp = FastMCP("我的第一个服务器")


@mcp.tool()
def add(a: float, b: float) -> float:
    """把两个数字相加。

    文档字符串会成为工具描述，类型标注会成为参数 schema——
    模型靠它们来决定何时、如何调用这个工具，务必认真写。
    """
    return a + b


@mcp.tool()
def greet(name: str, formal: bool = False) -> str:
    """向某人打招呼。

    Args:
        name: 对方的名字
        formal: 是否使用正式语气
    """
    if formal:
        return f"您好，{name}，很高兴为您服务。"
    return f"嗨，{name}！"


if __name__ == "__main__":
    # stdio 传输：客户端会把本脚本作为子进程拉起
    mcp.run()

import asyncio
import random
from typing import Annotated

from pydantic import BaseModel, Field
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError

mcp = FastMCP("进阶工具示例")


# ---------- 1. 参数校验：用 Annotated + Field 写清约束 ----------

@mcp.tool()
def divide(
    a: float,
    b: Annotated[float, Field(description="除数，不能为 0")],
    precision: Annotated[int, Field(ge=0, le=10, description="保留小数位数")] = 2,
) -> float:
    """计算 a 除以 b 的结果。"""
    if b == 0:
        # 抛 ToolError：错误信息会以 isError=true 的形式返回给模型，
        # 模型能看到原因并自行修正参数重试。
        raise ToolError("除数不能为 0")
    return round(a / b, precision)


# ---------- 2. 结构化输出：返回 dict / Pydantic 模型 ----------

class WeatherReport(BaseModel):
    city: str
    temperature_c: float
    condition: str
    humidity_pct: int


@mcp.tool()
def get_weather(city: str) -> WeatherReport:
    """获取指定城市的当前天气（演示用假数据）。

    返回 Pydantic 模型时，FastMCP 会同时生成：
    - structuredContent：结构化 JSON，供程序消费
    - content：文本摘要，供模型阅读
    """
    # 真实项目里这里应该调天气 API，例如 httpx.get(...)
    return WeatherReport(
        city=city,
        temperature_c=round(random.uniform(-5, 35), 1),
        condition=random.choice(["晴", "多云", "小雨", "雷阵雨"]),
        humidity_pct=random.randint(20, 95),
    )


# ---------- 3. 异步工具：IO 密集操作务必用 async ----------

@mcp.tool()
async def fetch_url_title(url: str) -> str:
    """抓取网页标题（演示异步写法，用 asyncio.sleep 模拟网络请求）。"""
    if not url.startswith(("http://", "https://")):
        raise ToolError(f"非法 URL：{url}")
    await asyncio.sleep(0.5)  # 真实场景：resp = await httpx.AsyncClient().get(url)
    return f"《{url} 的标题》"


# ---------- 4. 工具注解（annotations）：告诉模型工具的副作用 ----------

@mcp.tool(annotations={"readOnlyHint": True, "openWorldHint": False})
def get_server_time() -> str:
    """返回服务器当前时间。只读工具，模型可以更放心地自动调用。"""
    from datetime import datetime
    return datetime.now().isoformat(timespec="seconds")


@mcp.tool(annotations={"destructiveHint": True})
def delete_all_data(confirm: bool = False) -> str:
    """清空全部数据（演示用）。destructiveHint 会提示客户端先让用户确认。"""
    if not confirm:
        raise ToolError("危险操作：请将 confirm 设为 true 以确认删除")
    return "（演示）所有数据已删除"


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

"""例程三：Context 上下文 —— 日志、进度报告、读取自身资源。"""
import asyncio

from fastmcp import FastMCP, Context

mcp = FastMCP("上下文示例")


# 先注册一个资源，供下面的工具演示 ctx.read_resource()
@mcp.resource("config://settings")
def get_settings() -> str:
    return '{"max_retry": 3, "timeout_s": 30}'


@mcp.tool()
async def process_files(file_count: int, ctx: Context) -> str:
    """模拟批量处理文件：演示进度报告 + 日志推送到客户端。

    Args:
        file_count: 要处理的文件数量（1-20）
    """
    file_count = max(1, min(file_count, 20))

    # 1) 日志：推送到客户端显示（不会污染 stdio 协议通道）
    await ctx.info(f"开始处理 {file_count} 个文件")
    await ctx.debug("这是一条调试日志，客户端可选择性显示")

    # 2) 读配置资源：服务器内部也能消费自己注册的资源
    settings = await ctx.read_resource("config://settings")
    await ctx.debug(f"已加载配置：{settings.text}")

    for i in range(file_count):
        await asyncio.sleep(0.2)  # 模拟处理
        # 3) 进度报告：客户端需带 progressToken 才会收到
        await ctx.report_progress(progress=i + 1, total=file_count)
        if i == file_count // 2:
            await ctx.warning("已处理过半，一切正常")

    await ctx.info("全部处理完成")
    return f"成功处理 {file_count} 个文件"


# Elicitation（执行中向用户要输入）—— FastMCP 2/3 的写法如下：
#
# @mcp.tool()
# async def delete_project(project_id: str, ctx: Context) -> str:
#     ok = await ctx.elicit(f"确认删除项目 {project_id}？", response_type=bool)
#     ...
#
# ⚠️ FastMCP 4 已将交互式输入重构为协议级「多轮交互」机制
# （InputRequiredResult / SEP-2322），写法有变化，请以你所用版本的
# 官方文档为准：https://gofastmcp.com/servers/context

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

import json
from pathlib import Path

from fastmcp import FastMCP

mcp = FastMCP("资源示例")

# 模拟的用户数据库
USERS = {
    "u001": {"name": "张三", "role": "admin", "email": "zhangsan@example.com"},
    "u002": {"name": "李四", "role": "developer", "email": "lisi@example.com"},
}


# ---------- 1. 静态资源：固定 URI ----------

@mcp.resource("app://version")
def app_version() -> str:
    """应用版本信息。"""
    return "mcp-tutorial v1.0.0"


@mcp.resource("data://users/list", mime_type="application/json")
def list_users() -> str:
    """全部用户列表（JSON）。"""
    return json.dumps(list(USERS.keys()), ensure_ascii=False)


# ---------- 2. 资源模板：URI 里的 {参数} 会变成函数参数 ----------

@mcp.resource("user://{user_id}/profile", mime_type="application/json")
def get_user_profile(user_id: str) -> str:
    """按用户 ID 读取用户资料。"""
    user = USERS.get(user_id)
    if user is None:
        raise ValueError(f"用户不存在：{user_id}")
    return json.dumps({"id": user_id, **user}, ensure_ascii=False)


@mcp.resource("file://{path}")
def read_text_file(path: str) -> str:
    """读取项目目录下的文本文件（注意防路径穿越）。"""
    base = Path(__file__).parent.resolve()
    target = (base / path).resolve()
    # 安全：拒绝逃出基准目录的路径（FastMCP 4 默认也会拒绝 ".."）
    if not str(target).startswith(str(base)):
        raise ValueError("非法路径")
    return target.read_text(encoding="utf-8")


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

from fastmcp import FastMCP
from fastmcp.prompts import Message

mcp = FastMCP("提示词示例")


# ---------- 1. 返回字符串：最简单，自动作为一条 user 消息 ----------

@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """生成一段代码审查请求的提示词。"""
    return (
        f"请审查以下 {language} 代码，重点关注：\n"
        f"1. 正确性与边界条件\n2. 性能问题\n3. 可读性与命名\n4. 安全隐患\n\n"
        f"```{language}\n{code}\n```"
    )


# ---------- 2. 返回消息列表：构造多轮对话上下文 ----------

@mcp.prompt()
def debug_helper(error_message: str) -> list:
    """构造「资深工程师帮你排错」的多轮提示词。"""
    return [
        Message(
            role="user",
            content="你是一位资深工程师。我先描述问题，你再帮我分析。",
        ),
        Message(role="assistant", content="好的，请把错误信息和相关上下文发给我。"),
        Message(
            role="user",
            content=f"我遇到了这个错误：\n\n{error_message}\n\n"
                    "请分析可能的原因，并按可能性从高到低给出排查步骤。",
        ),
    ]


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

"""
功能：
  - 工具：新建笔记 / 搜索笔记 / 删除笔记 / 统计
  - 资源：note://{id} 读取单条笔记；notes://all 全部笔记
  - 提示词：summarize_notes 让模型总结笔记内容
存储：同目录下的 notes.db（SQLite，标准库自带，零依赖）
"""
import sqlite3
from datetime import datetime
from pathlib import Path

from fastmcp import FastMCP, Context
from fastmcp.exceptions import ToolError

DB_PATH = Path(__file__).parent / "notes.db"
mcp = FastMCP("我的笔记本")


# ---------- 存储层 ----------

def get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute(
        """CREATE TABLE IF NOT EXISTS notes (
               id INTEGER PRIMARY KEY AUTOINCREMENT,
               title TEXT NOT NULL,
               content TEXT NOT NULL,
               created_at TEXT NOT NULL
           )"""
    )
    return conn


def row_to_dict(row: sqlite3.Row) -> dict:
    return {"id": row["id"], "title": row["title"],
            "content": row["content"], "created_at": row["created_at"]}


# ---------- Tools ----------

@mcp.tool()
async def create_note(title: str, content: str, ctx: Context) -> dict:
    """新建一条笔记。

    Args:
        title: 笔记标题
        content: 笔记正文
    """
    if not title.strip():
        raise ToolError("标题不能为空")
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO notes (title, content, created_at) VALUES (?, ?, ?)",
            (title.strip(), content, datetime.now().isoformat(timespec="seconds")),
        )
        note_id = cur.lastrowid
    await ctx.info(f"笔记已创建，id={note_id}")
    return {"id": note_id, "message": f"笔记《{title}》已保存"}


@mcp.tool(annotations={"readOnlyHint": True})
def search_notes(keyword: str = "") -> list[dict]:
    """搜索笔记。keyword 为空时返回全部笔记（按时间倒序）。

    Args:
        keyword: 在标题和正文中搜索的关键词
    """
    with get_conn() as conn:
        if keyword:
            rows = conn.execute(
                "SELECT * FROM notes WHERE title LIKE ? OR content LIKE ? "
                "ORDER BY id DESC",
                (f"%{keyword}%", f"%{keyword}%"),
            ).fetchall()
        else:
            rows = conn.execute("SELECT * FROM notes ORDER BY id DESC").fetchall()
    return [row_to_dict(r) for r in rows]


@mcp.tool(annotations={"destructiveHint": True})
def delete_note(note_id: int) -> str:
    """按 ID 删除一条笔记。"""
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM notes WHERE id = ?", (note_id,))
    if cur.rowcount == 0:
        raise ToolError(f"笔记不存在：id={note_id}")
    return f"笔记 {note_id} 已删除"


@mcp.tool(annotations={"readOnlyHint": True})
def notes_stats() -> dict:
    """返回笔记总数和最新一条的标题。"""
    with get_conn() as conn:
        total = conn.execute("SELECT COUNT(*) FROM notes").fetchone()[0]
        latest = conn.execute(
            "SELECT title FROM notes ORDER BY id DESC LIMIT 1"
        ).fetchone()
    return {"total": total, "latest_title": latest[0] if latest else None}


# ---------- Resources ----------

@mcp.resource("notes://all", mime_type="application/json")
def all_notes_resource() -> str:
    """全部笔记（JSON 数组）。"""
    import json
    with get_conn() as conn:
        rows = conn.execute("SELECT * FROM notes ORDER BY id DESC").fetchall()
    return json.dumps([row_to_dict(r) for r in rows], ensure_ascii=False, indent=2)


@mcp.resource("note://{note_id}", mime_type="text/markdown")
def note_resource(note_id: str) -> str:
    """按 ID 读取单条笔记，渲染为 Markdown。"""
    with get_conn() as conn:
        row = conn.execute("SELECT * FROM notes WHERE id = ?",
                           (int(note_id),)).fetchone()
    if row is None:
        raise ValueError(f"笔记不存在：id={note_id}")
    return f"# {row['title']}\n\n> 创建时间：{row['created_at']}\n\n{row['content']}"


# ---------- Prompts ----------

@mcp.prompt()
def summarize_notes() -> str:
    """让模型读取全部笔记并生成摘要。"""
    return (
        "请通过 notes://all 资源读取我的全部笔记，然后：\n"
        "1. 按主题归类；\n"
        "2. 每类给出 3 句话以内的摘要；\n"
        "3. 列出可能需要跟进的待办事项。"
    )


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

import os

from fastmcp import FastMCP
from fastmcp.server.auth import StaticTokenVerifier  # 仅适合开发/内网演示

mcp = FastMCP(
    "HTTP 示例服务器",
    # 开发阶段用静态 Token 就够了；生产请换 JWTVerifier / OAuthProxy
    # （auth 只对 HTTP 传输生效，stdio 不需要）
    auth=StaticTokenVerifier(
        tokens={"dev-secret-token": {"client_id": "dev-user", "scopes": []}}
    ),
)


@mcp.tool()
def ping() -> str:
    """健康检查：返回 pong。"""
    return "pong"


@mcp.tool()
def echo(message: str) -> str:
    """原样返回你发来的消息。"""
    return f"你说：{message}"


if __name__ == "__main__":
    mcp.run(
        transport="http",
        host=os.environ.get("HOST", "127.0.0.1"),  # 本地调试绑回环地址
        port=int(os.environ.get("PORT", "8000")),
    )