pywss MCP Server

pip3 install pywss

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

# 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