AI编程新思路：把模型接进工作流（含MCP实践）

开场：我们今天不讲课，聊聊思路

目标内容：
写一个agent，能够实现将外部网站的视频的内容转化成提纲和正文。这个智能体要能够爬取制定网站或链接的信息，同时具备反爬取的处理机制。能够做内容的解析和清洗，能够利用工具提取正文、标题、段落和列表等您结构化信息，去除广告、导航栏、评论等无关的内容。文本清洗能够去除多余的空格、特殊自符和脚本，保证内容纯净。内容处理上能够实现分段归纳，按照段落和章节自动归纳每部分的主题句，形成分层提纲。对正文内容要按照逻辑顺序整理，必要时可进行语句的润色或简化。

大家好，今天很高兴能和大家一起聊聊AI。

我想先问大家一个问题：我们在用Copilot或者其他AI工具时，最常做的是什么？可能大多数时候是让它帮我们写个函数、补全一段代码，或者解释一下错误。这很棒，AI确实是一个效率超高的“助手”。

但今天，我不“讲解”某个工具怎么用，也不“科普”某个概念。我想和大家分享一个思路的转变：我们如何能将AI从一个简单的“代码助手”，升级为一个能与我们并肩作战、共同探索未知的“开发伙伴”？

这趟分享，更像是我自己最近的一些探索和思考之旅。我会分享我遇到的一些问题，以及我尝试用AI解决这些问题的思路。希望这个过程能给大家带来一些新的启发。

---

思路一：工具太多，如何打造自己的“瑞士军刀”？

困惑点：我们真的缺AI工具吗？

Copilot、Claude、Cursor… 新工具层出不穷，每个看起来都很强大。但问题也来了：我是不是每个都要学？它们之间到底有什么区别？我今天用这个，明天用那个，感觉很混乱，效率反而低了。

我的探索思路：从“有什么”到“我要什么”

我意识到，问题的关键不是去追逐每一个新工具，而是先想清楚“我需要AI帮我做什么”。我把我的需求分成了几类：

1. “体力活”: 写那些烦人的、重复的样板代码。
2. “脑力活”: 帮我阅读和理解一个陌生的、巨大的代码库；或者在我遇到一个棘手的bug时，提供一些不同的分析角度。
3. “探索活”: 当我想学习一个全新的技术（比如鸿蒙、或者某个新的前端框架）时，有没有办法让AI成为我的专属导师，带我快速入门？

带着这些问题，我重新审视了那些工具，发现它们各自的“人设”就清晰了：

• Copilot: 集成在IDE里的“高效实习生”，擅长补全与样板代码加速；它是“产品编排+模型能力”的组合。
• 代码模型家族: 提供底层代码理解/生成能力，可通过API编排复杂任务（如跨语言迁移、正则解释、重构建议）。与IDE内产品的关系是“模型能力”与“产品编排”的分工，而非简单的“谁是谁的幕后”。
• codex,kilo,Augment，roo…等等很多
• Claude: 更像经验丰富的“架构师”，长上下文阅读与审阅大型代码库见长；生成实现时建议配合工具接入与测试约束。
• Cursor 和支持知识挂载/工作流的工具: 提供“AI改装车间”，可挂载私有知识库（RAG）、接入工具/工作流（如MCP类机制），以胜任更垂直的“探索活”。

核心思路：MCP——让工具“活”起来的魔法

在探索中我发现了一个关键的东西：模型上下文协议（MCP）。

简单来说，它就是给AI装上“手、脚、眼睛、耳朵”的一套标准。它让AI不再是一个只能聊天的“大脑”，而是可以实际地去调用工具（Tool）、访问资源（Resource）。

一句话定义：MCP是一套标准化协议，规范模型如何安全地调用外部工具与访问资源，从而把“对话”接入“工作流”。

关键词：工具（Tool）、资源（Resource）、权限与安全（Auth/Scope）。

有了MCP，我们就可以把上面的思路变成现实。这不再是理论，而是可以立即上手的实践。比如：

• 场景1：AI变身项目管家

• 困惑点：每天在IDE和GitHub之间来回切换，处理Issue、看PR，很琐碎。

  • MCP思路：为AI装上 github 工具。现在，我可以直接在IDE里对AI说：“嘿，帮我列出所有带有‘bug’标签的紧急Issue”，或者“给第#123号PR加上一条评论：‘这里的逻辑似乎没有考虑到并发情况’”。AI就变成了我的项目管家。
  • 注意点：权限最小化（只读/仓库范围）、速率限制（API配额）、合规（避免输出敏感信息/密钥）。

• 场景2：AI变身技术专家

• 困惑点：要用的某个库更新了API，或者要接触一个全新的技术栈，文档看得头大。

  • MCP思路：为AI装上 context7 工具，然后把这个库的最新官方文档“喂”给它。现在，我可以问它：“在新版API中，那个废弃的oldFunction()现在应该用什么代替？” AI会基于最新文档，给出最权威的答案。
  • 注意点：标注文档版本与来源、避免版权风险、在回答中返回溯源片段与链接。

我的感悟是： 不要被工具绑架。先定义你的问题，再去组合你的工具。MCP就是那个能让你自由组合、打造属于自己独一无二“AI瑞士军刀”的关键。

---

思路二：AI能为我写代码，那“我”还能做什么？

困惑点：当AI越来越强，我们的价值在哪里？

这是一个有点焦虑但又必须面对的问题。如果AI能完成大部分编码工作，我们开发者的角色会变成什么？

我的探索思路：从“执行者”转变为“指挥家”

我尝试把自己定位成一个“项目经理”或“AI Agent的指挥家”。我的核心工作不再是逐行写代码，而是三件事：

1. 分解任务 (Decomposition): 把一个模糊的需求，拆解成AI能理解的、清晰的、可执行的步骤。这极其考验我们对业务和架构的理解能力。
2. 定义接口 (Interface Definition): 如果我让一个“前端AI”和一个“后端AI”同时工作，我必须先定义好它们之间沟通的“API契约”。我定义接口的质量，直接决定了AI协作的效率。
3. 监督与验证 (Supervision & Verification): AI会犯错。它可能会“创造性”地写出一些有问题的代码。我的价值在于，建立一套机制去自动化地验证它的工作，比如强制它为自己写的代码生成单元测试，并建立“AI代码审查AI”的流程。

案例分享：我是如何“教会”AI开发鸿蒙应用的

这个思路最好的验证，就是去挑战一个我完全不懂的领域。

我的做法，也正是我想分享的核心思路——“监督式AI开发”：

步骤清单（可落地）：

1. 确立“唯一事实来源”(SSOT)：收集官网最新的官方文档/API手册/规范，形成可版本化的知识库。
2. 授权与约束：通过知识挂载（RAG等）授权给AI，并下达硬约束：“仅允许基于知识库回答，缺失时回答‘不知道’并请求补充。”
3. 任务驱动问答：按“需求→接口→实现→测试”的顺序发起请求，要求给出依据章节/链接。
4. 验证与回归：强制生成单元测试/集成检查；失败即回溯来源，必要时补充知识库并重试。

失败样例与纠错流程：

• 情况：回答超出知识库或出现编造。
• 处置：拒绝采用→要求返回依据→补充/更新知识库→重试并记录溯源→纳入回归用例。

精髓：开发者掌握“解释权与验证权”，AI成为遵循“法律”的高效执行者。

实效：以此方式完成项目创建、编写与调试，并可对每段实现追溯到具体文档章节。

角色转变：开发者更像AI训练师/项目监督者，定义边界、制定契约、执行验证。

1. 确立“唯一事实来源” (Single Source of Truth): 我不去网上零散地搜索教程，而是直接从鸿蒙开发者官网，把最新的、最权威的官方文档、API手册、开发规范下载下来。这就是我们接下来所有工作的“法律”。

2. “授权”与“约束”AI: 我使用支持挂载文档的工具（其核心就是Augment或RAG技术），将这个“法律”文件集（知识库）授权给AI。同时，我下达一个绝对指令：“你现在是一个鸿蒙开发专家，你接下来所有的回答和代码，都必须、也只能来源于我提供的这份知识库。如果知识库里没有，就回答‘不知道’，绝不允许你自由发挥。”

3. 我来提问，AI来回答和实现:

   • 我：“根据你学习的文档，创建一个最基础的鸿蒙项目。”
   • AI：“好的，根据《应用开发指南》3.1节，您需要执行以下命令… 这是生成的项目结构…”
   • 我：“很好，现在根据《ArkTS开发范式》第5章，帮我创建一个包含按钮和文本的简单页面。”
   • AI：“遵命。这是符合规范的代码… 其中@Entry装饰器的作用是…”

这个思路的精髓在于：我，作为开发者，掌握了最终的“解释权”和“验证权”。AI则从一个不可控的“黑盒”，变成了一个严格遵守我所提供的“法律”的、高效的“执行者”。

结果是惊人的： AI真的像一个专家一样，一步步引导我完成了项目创建、代码编写和调试。当它生成的代码有疑问时，我会追问：“你这段代码是基于哪篇文档的第几节？” 它能准确地溯源。

我的感悟是： 在这个过程中，我的角色不是鸿蒙开发者，而是一个**“AI训练师”和“项目监督者”**。我定义了知识的边界（官方文档），并验证了结果的正确性。这可能就是我们未来的一种核心价值。

---

常见误解与坑

• 把MCP当成某个单一产品，而非一类协议/机制。
• 认为RAG能解决所有知识问题，忽视来源质量与版权。
• 误以为“上下文越长越好”，忽略检索质量与指令清晰度。
• 觉得“AI够强就不需要测试”，忽略监督与验证的重要性。

行动清单（会后即刻实践）

1. 明确你要解决的3类问题（体力/脑力/探索）。
2. 选择一个场景，先小范围试点。
3. 准备SSOT：下载并整理权威文档。
4. 搭建知识挂载与工具接入（如文档索引+最小工具）。
5. 写下硬约束与验收标准（必须溯源、必须有测试）。
6. 用“需求→接口→实现→测试”的提示模板驱动对话。
7. 对关键回答要求“出处+片段”。
8. 记录失败案例并加入回归集。
9. 度量时间节省与缺陷率，复盘迭代。
10. 将成功方案产品化/流程化，推广到团队。

流程图：监督式AI开发（SSOT→约束→实现→验证）

MCP推荐合集（实践选型）

• 核心基础能力
  • tavily：网页搜索与检索增强（npx tavily-mcp@0.2.3）
  • context7：技术文档问答/挂载官方文档（npx @upstash/context7-mcp@latest）
  • mcp-deepwiki：维基深度搜索（npx mcp-deepwiki@latest）
• 交互与执行
  • github：Issue/PR查询与评审（ghcr.io/github/github-mcp-server）
  • playwright：浏览器自动化与E2E测试（npx @playwright/mcp@latest）
• 任务管理与反馈
  • shrimp-task-manager：任务分解与跟踪（npx mcp-shrimp-task-manager）
  • mcp-feedback-enhanced：增强交互式反馈（uvx mcp-feedback-enhanced@latest）
• 通用能力
  • sequential-thinking：链式思考（npx @modelcontextprotocol/server-sequential-thinking）
  • memory：持久记忆（npx @modelcontextprotocol/server-memory，MEMORY_FILE_PATH=./memory.json）
  • mcp-server-time：统一时间信息（uvx mcp-server-time --local-timezone=Asia/Shanghai）

总结：我们的新角色——AI时代的“数字工匠”

聊了这么多，我想表达的核心思路是：AI时代，我们开发者的价值发生了转移。

• 从代码的生产者，转变为高质量指令的提供者。
• 从技术的实现者，转变为复杂问题的分解者和AI能力的整合者。
• 从被动学习者，转变为主动的“AI训练师”。

我们更像一个“数字工匠”，我们手里的工具不再是锤子和钉子，而是各种强大的AI Agent。而我们的核心技艺，就是如何构思、指挥、并组合这些AI，去创造出优秀的作品——并以“出处可追溯、结果可验证”的方式把AI接入真实工作流。

这趟探索之旅才刚刚开始，希望今天的分享能为大家打开一扇新的大门。
所以回归开始的话题内容，怎么做呢，这是我选用的是blibli的mcp工具，可以直接读取视频内容，然后通过这个归纳总结。
算是回归开头的一个需求总结。
bcut_asr.py：负责与 B站必剪语音识别接口通信。

import requests
import logging
import sys
import time
from os import PathLike
from pathlib import Path
from typing import Literal, Optional
from enum import Enum
from pydantic import BaseModel
import re
from urllib.parse import urlparse, parse_qs

INFILE_FMT = ["flac", "aac", "m4a", "mp3", "wav"]
OUTFILE_FMT = ["srt", "json", "lrc", "txt"]

__version__ = "0.0.3"

API_BASE_URL = "https://member.bilibili.com/x/bcut/rubick-interface"

# 申请上传
API_REQ_UPLOAD = API_BASE_URL + "/resource/create"

# 提交上传
API_COMMIT_UPLOAD = API_BASE_URL + "/resource/create/complete"

# 创建任务
API_CREATE_TASK = API_BASE_URL + "/task"

# 查询结果
API_QUERY_RESULT = API_BASE_URL + "/task/result"

SUPPORT_SOUND_FORMAT = Literal["flac", "aac", "m4a", "mp3", "wav", "mp4", "m4s"]

INFILE_FMT = ["flac", "aac", "m4a", "mp3", "wav", "mp4", "m4s"]
OUTFILE_FMT = ["srt", "json", "lrc", "txt"]

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
    "Cache-Control": "no-cache"
}

def get_audio_subtitle(url: str):
    asr = BcutASR(file=url)
    try:
        task_id = asr.create_task()
        while True:
            task_resp = asr.result(task_id)
            match task_resp.state:
                case ResultStateEnum.ERROR:
                    return APIError(task_resp.code, task_resp.msg)
                case ResultStateEnum.COMPLETE:
                    return task_resp.parse().to_txt()
                
            time.sleep(5)
    except Exception as e:
        return APIError(400, str(e) or "获取音频字幕失败")

class APIError(Exception):
    "接口调用错误"

    def __init__(self, code, msg) -> None:
        self.code = code
        self.msg = msg
        super().__init__()

    def __str__(self) -> str:
        return f"{self.code}:{self.msg}"


class ASRDataSeg(BaseModel):
    """文字识别-断句"""

    class ASRDataWords(BaseModel):
        """文字识别-逐字"""

        label: str
        start_time: int
        end_time: int

    start_time: int
    end_time: int
    transcript: str
    words: list[ASRDataWords]

    def to_srt_ts(self) -> str:
        """转换为srt时间戳"""

        def _conv(ms: int) -> tuple[int, int, int, int]:
            return ms // 3600000, ms // 60000 % 60, ms // 1000 % 60, ms % 1000

        s_h, s_m, s_s, s_ms = _conv(self.start_time)
        e_h, e_m, e_s, e_ms = _conv(self.end_time)
        return f"{s_h:02d}:{s_m:02d}:{s_s:02d},{s_ms:03d} --> {e_h:02d}:{e_m:02d}:{e_s:02d},{e_ms:03d}"

    def to_lrc_ts(self) -> str:
        """转换为lrc时间戳"""

        def _conv(ms: int) -> tuple[int, int, int]:
            return ms // 60000, ms // 1000 % 60, ms % 1000 // 10

        s_m, s_s, s_ms = _conv(self.start_time)
        return f"[{s_m:02d}:{s_s:02d}.{s_ms:02d}]"


class ASRData(BaseModel):
    """语音识别结果"""

    utterances: list[ASRDataSeg]
    version: str

    def __iter__(self):
        return iter(self.utterances)

    def has_data(self) -> bool:
        """是否识别到数据"""
        return len(self.utterances) > 0

    def to_txt(self) -> str:
        """转成 txt 格式字幕 (无时间标记)"""
        return "".join(seg.transcript for seg in self.utterances)

    def to_srt(self) -> str:
        """转成 srt 格式字幕"""
        return "\n".join(
            f"{n}\n{seg.to_srt_ts()}\n{seg.transcript}\n"
            for n, seg in enumerate(self.utterances, 1)
        )

    def to_lrc(self) -> str:
        """转成 lrc 格式字幕"""
        return "\n".join(
            f"{seg.to_lrc_ts()}{seg.transcript}" for seg in self.utterances
        )

    def to_ass(self) -> str:
        """转换为 ass 格式"""
        # TODO: ass 序列化实现
        raise NotImplementedError


class ResourceCreateRspSchema(BaseModel):
    """上传申请响应"""

    resource_id: str
    title: str
    type: int
    in_boss_key: str
    size: int
    upload_urls: list[str]
    upload_id: str
    per_size: int


class ResourceCompleteRspSchema(BaseModel):
    """上传提交响应"""

    resource_id: str
    download_url: str


class TaskCreateRspSchema(BaseModel):
    """任务创建响应"""

    resource: str
    result: str
    task_id: str  # 任务id


class ResultStateEnum(Enum):
    """任务状态枚举"""

    STOP = 0  # 未开始
    RUNING = 1  # 运行中
    ERROR = 3  # 错误
    COMPLETE = 4  # 完成


class ResultRspSchema(BaseModel):
    """任务结果查询响应"""

    task_id: str  # 任务id
    result: Optional[str] = None  # 结果数据-json
    remark: str  # 任务状态详情
    state: ResultStateEnum  # 任务状态

    def parse(self) -> ASRData:
        "解析结果数据"
        return ASRData.model_validate_json(self.result)


class BcutASR:
    "必剪 语音识别接口"
    session: requests.Session
    sound_name: str
    sound_url: str
    sound_bin: bytes
    sound_fmt: SUPPORT_SOUND_FORMAT
    __in_boss_key: str
    __resource_id: str
    __upload_id: str
    __upload_urls: list[str]
    __per_size: int
    __clips: int
    __etags: list[str]
    __download_url: str
    task_id: str

    def __init__(self, file: Optional[str | PathLike] = None) -> None:
        self.session = requests.Session()
        self.task_id = None
        self.__etags = []
        if file:
            self.set_data(file)

    def set_data(
        self,
        file: Optional[str | PathLike] = None,
        raw_data: Optional[bytes] = None,
        data_fmt: Optional[SUPPORT_SOUND_FORMAT] = None,
    ) -> None:
        "设置欲识别的数据"
        if file:
            if not isinstance(file, (str, PathLike)):
                raise TypeError("unknow file ptr")
            
            if re.match(r'^https?://', file):
                self.sound_url = file
                # 使用 urlparse 解析 URL
                parsed_url = urlparse(file)
                
                # 获取后缀名的优先级：
                # 1. 用户指定的 data_fmt
                # 2. URL 路径中的后缀
                # 3. 默认使用 m4s（B站音频格式）
                if data_fmt:
                    suffix = data_fmt
                else:
                    path = parsed_url.path.split('/')[-1]  # 获取文件名部分
                    if '.' in path:
                        suffix = path.split('.')[-1]
                    else:
                        suffix = 'm4s'  # B站默认音频格式
                
                self.sound_name = f'audio.{suffix}'
                self.__download_url = file
            else:
                # 文件类
                file = Path(file)
                self.sound_bin = open(file, "rb").read()
                suffix = data_fmt or file.suffix[1:]
                self.sound_name = file.name
        elif raw_data:
            # bytes类
            self.sound_bin = raw_data
            suffix = data_fmt
            self.sound_name = f"{int(time.time())}.{suffix}"
        else:
            raise ValueError("none set data")
        if suffix not in SUPPORT_SOUND_FORMAT.__args__:
            raise TypeError(f"format {suffix} is not support")
        self.sound_fmt = suffix
        logging.info(f"加载文件成功: {self.sound_name}")

    def upload(self) -> None:
        "申请上传"
        if not self.sound_bin or not self.sound_fmt or not self.sound_url:
            raise ValueError("none set data")
        resp = self.session.post(
            API_REQ_UPLOAD,
            data={
                "type": 2,
                "name": self.sound_name,
                "size": len(self.sound_bin),
                "resource_file_type": self.sound_fmt,
                "model_id": '8',
            },
            headers=headers,
        )
        resp.raise_for_status()
        resp = resp.json()
        code = resp["code"]
        if code:
            raise APIError(code, resp["message"])
        resp_data = ResourceCreateRspSchema.parse_obj(resp["data"])
        self.__in_boss_key = resp_data.in_boss_key
        self.__resource_id = resp_data.resource_id
        self.__upload_id = resp_data.upload_id
        self.__upload_urls = resp_data.upload_urls
        self.__per_size = resp_data.per_size
        self.__clips = len(resp_data.upload_urls)
        logging.info(
            f"申请上传成功, 总计大小{resp_data.size // 1024}KB, {self.__clips}分片, 分片大小{resp_data.per_size // 1024}KB: {self.__in_boss_key}"
        )
        self.__upload_part()
        self.__commit_upload()

    def __upload_part(self) -> None:
        "上传音频数据"
        for clip in range(self.__clips):
            start_range = clip * self.__per_size
            end_range = (clip + 1) * self.__per_size
            logging.info(f"开始上传分片{clip}: {start_range}-{end_range}")
            resp = self.session.put(
                self.__upload_urls[clip],
                data=self.sound_bin[start_range:end_range],
                headers=headers,
            )
            resp.raise_for_status()
            etag = resp.headers.get("Etag")
            self.__etags.append(etag)
            logging.info(f"分片{clip}上传成功: {etag}")

    def __commit_upload(self) -> None:
        "提交上传数据"
        resp = self.session.post(
            API_COMMIT_UPLOAD,
            data={
                "in_boss_key": self.__in_boss_key,
                "resource_id": self.__resource_id,
                "etags": ",".join(self.__etags),
                "upload_id": self.__upload_id,
                "model_id": "8",
            },
            headers=headers,
        )
        resp.raise_for_status()
        resp = resp.json()
        code = resp["code"]
        if code:
            raise APIError(code, resp["message"])
        resp_data = ResourceCompleteRspSchema.model_validate(resp["data"])
        self.__download_url = resp_data.download_url
        logging.info(f"提交成功")

    def create_task(self) -> str:
        "开始创建转换任务"
        resp = self.session.post(
            API_CREATE_TASK, json={"resource": self.__download_url, "model_id": "8"},
            headers=headers,
        )
        resp.raise_for_status()
        resp = resp.json()
        code = resp["code"]
        if code:
            raise APIError(code, resp["message"])
        resp_data = TaskCreateRspSchema.model_validate(resp["data"])
        self.task_id = resp_data.task_id
        logging.info(f"任务已创建: {self.task_id}")
        return self.task_id

    def result(self, task_id: Optional[str] = None) -> ResultRspSchema:
        "查询转换结果"
        resp = self.session.get(
            API_QUERY_RESULT, params={"model_id": "8", "task_id": task_id or self.task_id},headers=headers,
        )
        resp.raise_for_status()
        resp = resp.json()
        code = resp["code"]
        if code:
            raise APIError(code, resp["message"])
        return ResultRspSchema.model_validate(resp["data"])

bilibili_api：第三方库，负责调用 B站官方 API 获取视频信息和搜索。
server.py: 代码核心运行文件

import asyncio
import os
import aiohttp
from datetime import datetime
from tabulate import tabulate
from bilibili_api import video, Credential, search
from mcp.server.fastmcp import FastMCP
from bcut_asr import get_audio_subtitle

# 从环境变量获取认证信息
SESSDATA = os.getenv('sessdata')
BILI_JCT = os.getenv('bili_jct')
BUVID3 = os.getenv('buvid3')

# 初始化 Credential
credential = Credential(sessdata=SESSDATA, bili_jct=BILI_JCT, buvid3=BUVID3)

mcp = FastMCP("bilibili-mcp")

@mcp.tool("search_video", description="搜索bilibili视频")
async def search_video(keyword: str, page: int = 1, page_size: int = 20) -> str:
    """
    keyword: 搜索关键词
    page: 页码，默认1
    page_size: 每页数量，默认20
    """
    search_result = await search.search_by_type(keyword, search_type=search.SearchObjectType.VIDEO, page=page, page_size=page_size)
    
    # 准备表格数据
    table_data = []
    headers = ["发布日期", "标题", "UP主", "时长", "播放量", "点赞数", "类别", "bvid"]
    
    for video in search_result["result"]:
        # 转换发布时间
        pubdate = datetime.fromtimestamp(video["pubdate"]).strftime("%Y/%m/%d")
        
        # 将标题转换为Markdown链接格式
        title_link = f"[{video['title']}]({video['arcurl']})"
        
        table_data.append([
            pubdate,
            title_link,
            video["author"],
            video["duration"],
            video["play"],
            video["like"],
            video["typename"],
            video["bvid"]
        ])
    
    # 使用 tabulate 生成 Markdown 表格
    return tabulate(table_data, headers=headers, tablefmt="pipe")

@mcp.tool("get_video_subtitle", description="获取bilibili视频的字幕，需提供视频BV号")
async def get_video_subtitle(bvid: str) -> dict:
    """
    bvid: 视频BV号
    """
    v = video.Video(bvid=bvid, credential=credential)
    cid = await v.get_cid(page_index=0)
    info = await v.get_player_info(cid=cid)
    json_files = info["subtitle"]["subtitles"]
    
    # 过滤查找符合条件的字幕
    target_subtitle = None
    for subtitle in json_files:
        if subtitle["lan"] == "ai-zh" and subtitle["lan_doc"] == "中文（自动生成）":
            target_subtitle = subtitle
            break
    
    if not target_subtitle:
        # 检查视频时长，如果超过20分钟则不进行ASR
        video_info = await v.get_info()
        duration = video_info.get('duration', 0)
        if duration > 1200: # 20分钟 = 1200秒
            return "视频时长超过20分钟，无法生成实时字幕。"

        url_res = await v.get_download_url(cid=cid)
        # 提取音频URL
        audio_arr = url_res.get('dash', {}).get('audio', [])
        if not audio_arr:
            return "没有找到AI生成的中文字幕"
            
        # 获取最后一个音频（通常质量最高）
        audio = audio_arr[-1]
        # 选择合适的音频URL
        audio_url = ""
        if '.mcdn.bilivideo.cn' in audio['baseUrl']:
            audio_url = audio['baseUrl']
        else:
            backup_url = audio.get('backupUrl', [])
            if backup_url and 'upos-sz' in backup_url[0]:
                audio_url = audio['baseUrl']
            else:
                audio_url = backup_url[0] if backup_url else audio['baseUrl']
        
        asr_data = get_audio_subtitle(audio_url)
        return asr_data
    
    # 确保 URL 有 HTTPS 前缀
    subtitle_url = target_subtitle["subtitle_url"]
    if not subtitle_url.startswith(('http://', 'https://')):
        subtitle_url = f"https:{subtitle_url}"
    
    # 使用 aiohttp 获取字幕内容
    async with aiohttp.ClientSession() as session:
        async with session.get(subtitle_url) as response:
            subtitle_content = await response.json()
            # 提取并拼接所有字幕文本
            if "body" in subtitle_content:
                subtitle_text = "".join(item["content"] for item in subtitle_content["body"])
                return subtitle_text
            else:
                return subtitle_content

@mcp.tool("get_video_info", description="获取bilibili视频信息，需提供视频BV号")
async def get_video_info(bvid: str) -> dict:
    """
    bvid: 视频BV号
    """
    v = video.Video(bvid=bvid, credential=credential)
    info = await v.get_info()
    return info

@mcp.tool("get_media_subtitle", description="获取媒体文件的AI中文字幕，需提供媒体文件URL")
async def get_media_subtitle(url: str) -> dict:
    """
    url: 媒体文件URL
    """
    asr_data = get_audio_subtitle(url)
    return asr_data

mcp.run()

然后需要安装的文件主要 requirements.txt

mcp
bilibili-api-python
aiohttp

这三个核心文件来运行和安装

其中操作是

1. 安装依赖
   首先需要安装 Python 依赖包：

pip install -r requirements.txt

需要安装的包包括：

mcp - MCP SDK
bilibili-api-python - B站 API Python 库
aiohttp - 异步 HTTP 客户端/服务器框架
另外代码中还使用了 tabulate 和 pydantic，需要补充安装：

pip install tabulate pydantic

2. 配置 B 站认证信息
   需要从 B 站网站的 cookies 中获取以下三个参数并设置为环境变量：
   sessdata - B站 SESSDATA
   bili_jct - B站 bili_jct
   buvid3 - B站 buvid3
   获取方式：
   在浏览器中登录 B 站
   按 F12 打开开发者工具
   进入 Application/存储 → Cookies → https://www.bilibili.com
   找到对应的 cookie 值
   设置环境变量（Windows）：

set sessdata=你的SESSDATA值
set bili_jct=你的bili_jct值
set buvid3=你的buvid3值

3. 运行服务器

python server.py

4. 集成到桌面应用（可选）
   如果要将此 MCP 服务器集成到支持 MCP 的应用中，可以在应用的配置文件中添加：

{
  "mcpServers": {
    "bilibili-mcp": {
      "command": "python",
      "args": [
        "d:/mcpproject/bilibili-mcp/server.py(这个地方应该填写你得文件夹位置)"  
      ],
      "env": {
        "sessdata": "你的SESSDATA值",
        "bili_jct": "你的bili_jct值",
        "buvid3": "你的buvid3值"
      }
    }
  }
}

该服务器提供了 4 个工具：

search_video - 搜索 B 站视频
get_video_subtitle - 获取视频字幕（AI 自动生成）
get_video_info - 获取视频详细信息
get_media_subtitle - 使用必剪 API 为媒体文件生成 AI 字幕