server.py•5.04 kB
import asyncio
import logging
import os
from datetime import datetime
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio
from playwright_server.tools.handles import NavigateToolHandler, ScreenshotToolHandler, EvaluateToolHandler, GetTextContentToolHandler, GetHtmlContentToolHandler, NewSessionToolHandler
from playwright_server.tools.action import ActionToolHandler
# 配置日志记录
def setup_logging():
# 创建日志目录
log_dir = "logs"
if not os.path.exists(log_dir):
os.makedirs(log_dir)
# 创建带有时间戳的日志文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = os.path.join(log_dir, f"playwright_server_{timestamp}.log")
# 配置日志格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler(log_file)
]
)
logger = logging.getLogger('playwright_server')
logger.info(f"日志系统初始化完成，日志文件: {log_file}")
return logger
# 初始化日志系统
logger = setup_logging()
# 定义所有工具处理程序实例列表
tool_handler_list = [
NavigateToolHandler(),
# ScreenshotToolHandler(),
EvaluateToolHandler(),
GetTextContentToolHandler(),
GetHtmlContentToolHandler(),
NewSessionToolHandler(),
ActionToolHandler()
]
# 根据每个处理程序的 name 属性创建字典
tool_handlers = {handler.name: handler for handler in tool_handler_list}
server = Server("playwright-server")
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
"""
List available note resources.
Each note is exposed as a resource with a custom note:// URI scheme.
"""
logger.debug("处理 list_resources 请求")
return []
@server.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
"""
Read a specific note's content by its URI.
The note name is extracted from the URI host component.
"""
logger.debug(f"处理 read_resource 请求，URI: {uri}")
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
"""
List available prompts.
Each prompt can have optional arguments to customize its behavior.
"""
logger.debug("处理 list_prompts 请求")
return []
@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
"""
Generate a prompt by combining arguments with server state.
The prompt includes all current notes and can be customized via arguments.
"""
logger.debug(f"处理 get_prompt 请求，名称: {name}, 参数: {arguments}")
raise ValueError(f"Unknown prompt: {name}")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
logger.debug("处理 list_tools 请求")
return [
tool_handler.to_tool() for tool_handler in tool_handlers.values()
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Tools can modify server state and notify clients of changes.
"""
logger.info(f"处理 call_tool 请求，工具: {name}, 参数: {arguments}")
if name in tool_handlers:
try:
logger.debug(f"调用工具处理器: {name}")
result = await tool_handlers[name].handle(name, arguments)
logger.debug(f"工具处理器 {name} 执行完成")
return result
except Exception as e:
logger.error(f"工具处理器 {name} 执行失败: {str(e)}", exc_info=True)
raise
else:
logger.warning(f"未知工具: {name}")
raise ValueError(f"Unknown tool: {name}")
async def main():
# Run the server using stdin/stdout streams
logger.info("启动 Playwright 服务器")
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
try:
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="playwright-plus-server",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
except Exception as e:
logger.error(f"服务器运行出错: {str(e)}", exc_info=True)
if __name__ == "__main__":
asyncio.run(main())