Pywss 简介

Pywss（发音 /piːwaɪz/，类似 p~whys）是一个轻量级的 Python Web 框架，它基于 Python3.6+ 特性构建。

与 Flask、Django 等主流框架不同的是，Pywss 的底层并没有实现 WSGI 接口协议。 其编程风格也更类似于 Gin、Iris 等框架，因此对于熟悉这些框架的开发者来说，Pywss 是一个非常值得探索的项目。

其关键特性有：

• 简单：拒绝海量参数，减少心智负担。了解上下文 pywss.Context 即刻启程。
• 快速：引入线程池机制，减少并发场景下线程创建/销毁开销。
• 优雅：ctx.next 真的太优雅了。如果你也和我一样喜欢，那我觉得这件事情，泰裤辣！！
• 标准化：集成了部分 OpenAPI（Swagger）能力，方便开发者快速生成 API 文档并进行调试。
• 支持WebSocket：开箱即用的 WebSocket 能力。
• 接口测试：开箱即用的 API 测试模块，不启动服务也能测试接口功能辣！
• MCP PRO：一站式集成 SSE、StreamHTTP 和 MCPO 协议，助你轻松构建多 MCP 工具🔥

在线文档 https://czasg.github.io/pywss/_

快速开始

1、安装 pywss

pip3 install pywss

2、搭建 web 应用

首先创建 main.py 文件，并写入以下代码：

import time
import pywss
def log_handler(ctx: pywss.Context):     start_time = time.time()     ctx.next()     print(         f"Route: {ctx.route}, "         f"Method: {ctx.method}, "         f"Status: {ctx.responsestatuscode}, "         f"Time: {time.time() - start_time:.3f}s"     )
def handler(ctx: pywss.Context):   ctx.write("hello~")
def main():     app = pywss.App()     app.get("/hello", handler)  # curl localhost:8080/hello     app.any("*", log_handler, handler)  # curl -X POST localhost:8080/hello     app.run()
if name == 'main':     main()

接着启动服务:

python3 main.py

至此，一个简单的 web 应用服务就完成了。

3、搭建 MCP 应用

要快速构建 MCP 服务，只需继承 pywss.mcp.MCPServer 并遵循以下规则：

核心约束:

• 所有接口方法必须以 tool 开头（如 toolquery_user）
• 必须使用 @pywss.openapi.docs 声明请求参数，且 request 必须从 pydantic.BaseModel 继承（自动生成 OpenAPI 文档）

请求处理:

• 通过 ctx.data.req 直接获取结构化请求体
• 使用 self.handle_success(ctx, data) 返回成功响应
• 使用 self.handle_error(ctx, code, message) 返回标准错误

# coding: utf-8 import pywss from pydantic import BaseModel from pywss.mcp import MCPServer

class DomainReq(BaseModel):  # 定义 DomainReq 请求，必须从 pydantic.BaseModel 继承     domain: str
class DomainMCPServer(MCPServer):  # 定义 DomainMCPServer 服务，必须从 pywss.mcp.MCPServer 继承     @pywss.openapi.docs(description="获取单个域名服务", request=DomainReq)  # required，工具及其参数说明     def toolgetdomain(self, ctx: pywss.Context):         req: DomainReq = ctx.data.req  # 框架已经封装好了请求，可以从 ctx.data.req 直接获取使用，异常请求会被拦截         self.handlesuccess(ctx, {  # handlesuccess 封装了 jsonrpc2.0 输出规范             "domain": req.domain,             "color": req.color         })
class LogReq(BaseModel):     traceId: str
class LogMCPServer(MCPServer):     @pywss.openapi.docs(description="获取单个trace日志", request=LogReq)     def toolgettrace_log(self, ctx: pywss.Context):         req: LogReq = ctx.data.req         self.handle_success(ctx, {             "traceId": req.traceId,         })
domainMCPServer = DomainMCPServer() logMCPServer = LogMCPServer()
app = pywss.App() app.openapi()  # 开启 OpenAPI 文档 domainMCPServer.mount(app.group("/api/v1/domain"))  # 挂载 MCP 服务，同时指定路由 logMCPServer.mount(app.group("/api/v1/log"))  # 挂载 MCP 服务，同时指定路由 app.run()

接着启动服务:

python3 main.py

| 协议类型 | 服务类 | 请求方法 | 端点路径格式 | 示例路径 | |------------|----------------------|----------|---------------------------------------|---------------------------------------| | SSE | domainMCPServer | GET | /api/v1/domain/sse | GET /api/v1/domain/sse | | | logMCPServer | GET | /api/v1/log/sse | GET /api/v1/log/sse | | StreamHTTP | domainMCPServer | POST | /api/v1/domain/mcp | POST /api/v1/domain/mcp | | | logMCPServer | POST | /api/v1/log/mcp | POST /api/v1/log/mcp | | MCPO | domainMCPServer | POST | /api/v1/domain/tools/{toolname} | POST /api/v1/domain/tools/getdomain | | | logMCPServer | POST | /api/v1/log/tools/{toolname} | POST /api/v1/log/tools/gettrace_log |

更多功能见在线文档。

Activity