python-agent-development

Python Agent Development Skill (Windows)

Expert guidance for developing Windows Platform Agent using Python 3.12 + wxauto v4.

Project Structure

platform_agents/windows_agent/
├── agent.py                # 主入口
├── wechat_monitor.py      # 微信监听模块
├── input_writer.py        # 输入框控制模块
├── ipc/                   # IPC 通信模块
│   ├── message_sender.py
│   └── command_receiver.py
├── utils/                 # 工具模块
│   ├── logger.py
│   └── config.py
├── requirements.txt       # 依赖列表
├── tests/                 # 测试目录
│   ├── test_monitor.py
│   └── test_writer.py
└── README.md

Dependencies (requirements.txt)

wxauto==4.0.0
pywin32>=305

Main Entry Point

# agent.py
import sys
import time
from wechat_monitor import WeChatMonitor
from input_writer import WeChatInputWriter
from ipc.message_sender import MessageSender
from ipc.command_receiver import CommandReceiver

class WindowsAgent:
    def __init__(self):
        self.monitor = WeChatMonitor(interval_ms=500)
        self.input_writer = WeChatInputWriter()
        self.command_receiver = CommandReceiver()
        self.setup_command_handlers()

    def setup_command_handlers(self):
        """注册命令处理器"""
        self.command_receiver.register_handler("WriteInput", self.handle_write_input)
        self.command_receiver.register_handler("ClearInput", self.handle_clear_input)
        self.command_receiver.register_handler("HealthCheck", self.handle_health_check)
        self.command_receiver.register_handler("Stop", self.handle_stop)

    def handle_write_input(self, command):
        content = command.get('content', '')
        success = self.input_writer.write_to_input(content)
        return success

    def handle_clear_input(self, command):
        success = self.input_writer.clear_input()
        return success

    def handle_health_check(self, command):
        MessageSender.send_health_status("ok", "windows_wxauto")
        return None  # 不发送 CommandResponse

    def handle_stop(self, command):
        self.command_receiver.stop()
        return True

    def run(self):
        """启动 Agent"""
        try:
            # 启动命令监听（后台线程）
            self.command_receiver.start_listening_async()

            # 启动微信监听（主线程）
            self.monitor.start_monitoring()

        except KeyboardInterrupt:
            MessageSender.send_error("Agent 被用户中断")
        except Exception as e:
            MessageSender.send_error(f"Agent 运行错误: {str(e)}")
        finally:
            sys.exit(0)

if __name__ == '__main__':
    agent = WindowsAgent()
    agent.run()