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.response_status_code}, "
        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_ 开头（如 tool_query_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 tool_get_domain(self, ctx: pywss.Context):
        req: DomainReq = ctx.data.req  # 框架已经封装好了请求，可以从 ctx.data.req 直接获取使用，异常请求会被拦截
        self.handle_success(ctx, {  # handle_success 封装了 jsonrpc2.0 输出规范
            "domain": req.domain,
            "color": req.color
        })

class LogReq(BaseModel):
    traceId: str

class LogMCPServer(MCPServer):
    @pywss.openapi.docs(description="获取单个trace日志", request=LogReq)
    def tool_get_trace_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/{tool_name} | POST /api/v1/domain/tools/get_domain | | | logMCPServer | POST | /api/v1/log/tools/{tool_name} | POST /api/v1/log/tools/get_trace_log |

更多功能见在线文档。

Activity