实战篇:Python开发monogod数据库mcp server看完你就会了

实战篇:Python开发monogod数据库mcp server看完你就会了
原创不易,请关注公众号:【爬虫与大模型开发】,大模型的应用开发之路,整理了大模型在现在的企业级应用的实操及大家需要注意的一些AI开发的知识点!持续输出爬虫与大模型的相关文章。

前言

目前mcp协议是给deepseek大模型插上工具链的翅膀,让大模型不仅拥有超高的推理和文本生成能力,还能具备执行大脑意识的工具能力!

如何开发一个mcp?

mcp是一种协议,指的是模型上下文协议 (Model Context Protocol)。

官方结成的mcp
https://github.com/modelcontextprotocol/python-sdk

mcp库
pip install mcp from mcp.server.fastmcp import FastMCP 

我们先来做一个简单的案例

from mcp.server.fastmcp import FastMCP import requests mcp = FastMCP("spider") @mcp.tool() def crawl(url: str) -> int:     """获取网页的源代码""" resposne = requests.get(url)     return response.text if __name__ == "__main__":     mcp.run(transport='stdio') #mcp.run(transport="sse") 
以上代码实现了FastMCP 库来创建一个简单的 MCP 服务器,它提供一个基本的工具(工具名:crawl),用于是实现源代码的获取。这是一个非常简单的爬虫mcp入门示例,适合理解 MCP 服务器的工作原理和 FastMCP 的基本用法。

代码分析

首先,从 mcp.server.fastmcp 模块中导入了 FastMCP 类。接着,创建了一个 FastMCP 类的实例,命名为 spider。

FastMCP 是服务器的核心类,它负责管理工具、资源和通信。参数「spider」是服务器的名称,用于标识这个 MCP 服务器。

然后,@mcp.tool() 是一个装饰器(decorator),它告诉 FastMCP 将 crawl 函数注册为一个可供客户端调用的工具。

文档字符串(“”“获取网页的源代码”“”):这是函数的描述,客户端(如 MCP Inspector)会显示这个描述,帮助用户理解工具的功能。这个工具可以被外部客户端调用!

例如通过 MCP 协议发送请求,传入url,服务器会返回它的网页源代码。

mcp.run(transport=‘stdio’) 启动 MCP 服务器,进入监听状态,等待客户端的连接和请求。transport=‘stdio’ 指定了通信方式为标准输入输出(Standard I/O)。这意味着服务器通过命令行的标准输入(stdin)和标准输出(stdout)与客户端通信,适合本地开发和测试。

mcp.run(transport=‘stdio’)启动 MCP 服务器,如果需要远程通信,可以更改为其他传输方式(如 ‘sse’,Server-Sent Events)。

fastapi-mcp库
pip install fastapi-mcp from fastapi_mcp import add_mcp_server 

案例分析

import requests from fastapi import FastAPI, HTTPException, Query from pydantic import BaseModel from typing import Optional from fastapi_mcp import add_mcp_server # Create a simple FastAPI app app = FastAPI( title=spider", description="A simple example API with integrated MCP server", version="0.1.0", ) @app.get("/items/{url}", response_model=Item, tags=["items"]) def crawl(url: Optional[str] = Query(None, description="get websource from url")) -> dict:     """获取网页的源代码""" resposne = requests.get(url)     return {"response":response.text} mcp_server = add_mcp_server( app, mount_path="/mcp", # 挂着mcp服务器的地址 name="Item API MCP", # mcp服务的名称 description="MCP server for the Item API", base_url="http://localhost:8000", describe_all_responses=False, describe_full_response_schema=False, ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) 
fastapi_mcp分析

上面的案例引入了fastapi和fastapi-mcp并实例化一个fastapi的类;

然后使用fastapi编写了一个方法用户获取一个url站点的源代码;

使用fastspi-mcp创建一个实例化的mcp_server,将上面的fastapi服务注册为一个mcp服务,可以配置mcp服务调用的路由为http://localhost:8000/mcp

最后导入uvicorn库,正常启动fastapi的服务;可以无缝的将代码转成mcp server!

以上两种实现MCP服务的方式,都可以学习下,尤其是如果你本身对fastapi比较的熟练,那么可以直接使用fastapi-mcp将服务注册为MCP server!如果熟悉python,但是不熟悉fastapi也可以使用官方推荐的mcp库,节省学习的成本,降低学习难度系数!

实战编写mongodb数据库查询MCP服务

import sys from datetime import datetime import traceback import uvicorn from fastapi_mcp import add_mcp_server from fastapi import FastAPI, status, HTTPException, Response, Query import motor.motor_asyncio app = FastAPI() ......代码省略..... # 根据结构式获取化合物计算属性的值 @app.get("/api/computes/", status_code=status.HTTP_200_OK, summary="根据结构式获取化合物计算属性的值") async def computed(smiles: Optional[str] = Query(None, description="Search smiles string")): result = ChemRdkit(smiles=smiles).computed if result and result.get("code") == 200: del result["code"] return ResponseModel(data=result, message="success") return ErrorResponseModel(error=result.get("error"), code=status.HTTP_404_NOT_FOUND, message=result.get("message")) # 获取单个化合物 @app.get("/api/chemicals/{casno}/", status_code=status.HTTP_200_OK, summary="根据casno获取单个化合物") async def chemicals(casno:str): """" 根据casno获取化合物 """ result = await chemical_get_data(casno=casno) if result: return ResponseModel(data=result, message="success") return ErrorResponseModel(error="数据不存在", code=status.HTTP_404_NOT_FOUND, message="数据不存在") # raise HTTPException(400, "Something went wrong") # 访问结构式图片 @app.get("/image/{filename}/", status_code=status.HTTP_200_OK, summary="访问结构式图片") async def get_image(filename: str): print(filename) # 获取图片二进制 image_content = struct_obj.get_struct(filename) if image_content: return Response(content=image_content, media_type="image/jpeg") # 不存在,通过filename获取casno号的基本信息,使用rdkit库根据smiles(存在)生成结构式图片 obj = await get_smiles(casno=filename.replace("_", "-")) if obj: ChemRdkit(smiles=obj.get("smiles", "")).struct(filename=obj.get("casno")) image_content = struct_obj.get_struct(filename) if image_content: return Response(content=image_content, media_type="image/jpeg") return None mcp_server = add_mcp_server( app, mount_path="/mcp", # 挂着mcp服务器的地址 name="Item API MCP", # mcp服务的名称 description="MCP server for the Item API", base_url="http://192.168.1.250:9999", describe_all_responses=False, describe_full_response_schema=False, ) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=9999) 

结果输出

实现查询casno的基本信息,以及根据smiles生成化合物的属性信息

以上就可以把企业本地查询数据的一般能力集成到deepseek中了!让企业赶的上AI的快车!

搭建自己的本地MCP hub

Hub 服务器 (MCP Hub): 作为中央管理服务器,连接并管理多个 MCP 服务器。它为客户端提供统一的 API 入口,并将请求路由到相应的 MCP 服务器。

docker run --name=mcphub -p 3001:3000 --restart=always -d registry.cn-hangzhou.aliyuncs.com/samanhappy/mcphub 

浏览器输入:http://localhost:3001/

1panel MCP server

目前仅支持stdio的输出协议!

Ubuntu安装

curl -sSL https://resource.fit2cloud.com/1panel/package/quick_start.sh -o quick_start.sh && sudo bash quick_start.sh 
其他MCP服务参考

https://github.com/punkpeye/awesome-mcp-servers/tree/main

送书福利

DeepSeek作为开源大模型的典范,融合了Transformer架构、MoE(混合专家)机制及自监督学习等前沿技术,在性能与扩展性上优势显著。为助力技术人员系统掌握DeepSeek的开发原理与应用,笔者撰写了本书,旨在为读者提供从理论到实践的全面指导。

只要关注+点赞+评论 评论区选择2人送书

联系方式

加入星球:
原创不易,点个关注!
不会错过后面的优质文章!
觉着写的不错的可以帮忙点点赞
关注公众号:【爬虫与大模型开发】
需要以上源代码的下面关注并私信!
活跃在一线的Python工程师分享自己学习之路
我创建了【爬虫与大模型开发】的知识星球
适合爱好爬虫及从事爬虫的同学
需要代码提示词加我星球!

需要完整代码扫码星球自提!

Read more

【Linux】TCP可靠性与性能优化详解:从确认应答到拥塞控制

【Linux】TCP可靠性与性能优化详解:从确认应答到拥塞控制

文章目录 * TCP可靠性与性能优化详解:从确认应答到拥塞控制 * 一、确认应答(ACK)机制 * 1.1 什么是确认应答 * 1.2 序列号和确认号的作用 * 1.3 确认应答的图示 * 1.4 确认应答的问题 * 二、超时重传机制 * 2.1 为什么需要超时重传 * 2.2 超时重传的原理 * 2.3 超时重传的图示 * 2.4 超时时间如何确定 * 2.5 动态计算超时时间 * 2.6 重传次数限制 * 三、滑动窗口机制 * 3.1 滑动窗口的核心思想 * 3.2 滑动窗口的工作原理 * 3.3 滑动窗口的详细图示

By Ne0inhk

Ubuntu新手必看:如何快速更换国内源(阿里/清华/中科大源对比)

Ubuntu 新手的第一道“加速”关:国内镜像源深度解析与实战指南 刚装好 Ubuntu,那种清爽的桌面和开箱即用的感觉确实不错。但当你兴冲冲地打开终端,准备用 apt install 装点东西时,进度条那慢如蜗牛的爬行速度,是不是瞬间浇灭了一半的热情?别急着怀疑自己的网络,这几乎是每个国内 Ubuntu 用户都会遇到的“新手墙”。问题的核心,往往不在于你的宽带,而在于系统默认连接的软件仓库服务器远在海外,网络延迟和带宽限制成了最大的瓶颈。 解决这个问题的方法,就是“换源”——将系统的软件源地址,更换为位于国内的镜像服务器。这听起来像是个简单的操作,但背后其实有不少门道:国内有哪些可靠的镜像站?阿里云、清华大学、中国科学技术大学(USTC)的源有什么区别?为什么别人的源换上去飞快,你的却报了一堆错?今天,我们就来彻底拆解这个问题。这不仅仅是复制粘贴几行命令,而是帮你理解原理、掌握选择、并能在遇到问题时自己动手排查。无论你是刚接触 Linux 的开发新手,还是希望优化工作流效率的资深用户,一个配置得当的软件源,

By Ne0inhk
玩转ClaudeCode:ClaudeCode安装教程(Windows+Linux+MacOS)

玩转ClaudeCode:ClaudeCode安装教程(Windows+Linux+MacOS)

本文介绍如何安装 AI 编码界一骑绝尘的最强工具 ——— Claude Code。安装不同的操作系统环境,本文会从 Windows、Linux、Mac 三个不同的系统环境依次介绍安装方法。 其中,Windows 系统作为大家最主流的操作系统,提供了两种安装方式,一种方式是直接在 Windows 的终端里安装,另一种是在 Windows 的子系统(WSL)内完成安装。其中,通过 WSL 安装,我们又可以分为,WSL 环境的直装和基于 WSL 的容器化安装(Docker),几种方法各有利弊,但均可正常使用。 Windows 环境直装 Claude Code 1. 获取 Claude Code 账号 访问 Claude Code 中国镜像站,完成账户注册。 输入邀请码

By Ne0inhk
Flutter 三方库 swagger_parser 自动化打通鸿蒙 API 通信(一键将 Swagger 转化为 Dart 模型)

Flutter 三方库 swagger_parser 自动化打通鸿蒙 API 通信(一键将 Swagger 转化为 Dart 模型)

欢迎加入开源鸿蒙跨平台社区:https://openharmonycrossplatform.ZEEKLOG.net 前言 在进行 OpenHarmony 项目开发时,最枯燥的工作莫过于根据后端提供的 Swagger (OpenAPI) 文档手动编写一个个的 Request 类、Response 类和 API Client。这不仅低效,而且极易因文档更新没对齐而导致 Bug。 swagger_parser 是一个强大的命令行工具,它能直接读取本地或网络上的 Swagger JSON/YAML 文件,自动为你生成完整的 Dart 数据类和 Dio/Chopper API 控制器。 一、核心工作流 Swagger JSON / YAML Swagger Parser Dart 数据模型 (JSON Serialized) Dio / Chopper

By Ne0inhk