MCP与Skill开发入门：从零构建高效技能系统的实战指南

2次阅读

没有评论

共计 2725 个字符，预计需要花费 7 分钟才能阅读完成。

在游戏开发、自动化运维等场景中，我们经常遇到需要管理大量技能（Skill）的情况。传统实现方式通常存在以下问题：

状态管理混乱 ：技能执行状态分散在各处，难以追踪
并发控制薄弱 ：多个技能同时执行时容易引发资源竞争
缺乏统一抽象 ：不同技能的实现方式差异大，难以维护
执行效率低下 ：串行执行导致系统吞吐量受限

优点：

响应速度快
适合高频率触发的技能
天然支持异步处理

缺点：

状态追踪困难
错误处理复杂

优点：

状态流转清晰
容易实现事务性操作
便于调试和监控

缺点：

实现复杂度较高
性能开销稍大

选型建议 ：对于需要严格状态管理的业务场景，推荐使用状态机模型；对于简单的一次性技能，事件驱动更轻量。

class Skill:
    def __init__(self, skill_id, params=None):
        self.skill_id = skill_id
        self.params = params or {}
        self.status = "PENDING"  # PENDING, RUNNING, SUCCESS, FAILED, CANCELLED

    def execute(self, context):
        """核心执行方法"""
        self.status = "RUNNING"
        try:
            # 实际业务逻辑
            result = self._do_execute(context)
            self.status = "SUCCESS"
            return result
        except Exception as e:
            self.status = "FAILED"
            raise

    def cancel(self):
        """取消执行"""
        if self.status == "RUNNING":
            self._do_cancel()
        self.status = "CANCELLED"

    # 抽象方法，由具体技能实现
    def _do_execute(self, context):
        raise NotImplementedError

    def _do_cancel(self):
        pass

from concurrent.futures import ThreadPoolExecutor

class SkillExecutor:
    def __init__(self, max_workers=10):
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.running_tasks = {}

    def submit(self, skill, context):
        """提交技能执行"""
        if skill.skill_id in self.running_tasks:
            raise Exception("Skill already running")

        future = self.executor.submit(skill.execute, context)
        self.running_tasks[skill.skill_id] = future

        # 添加回调清理
        future.add_done_callback(lambda f: self.running_tasks.pop(skill.skill_id, None))
        return future

    def cancel(self, skill_id):
        """取消执行中的技能"""
        future = self.running_tasks.get(skill_id)
        if future and not future.done():
            future.cancel()
            return True
        return False

推荐采用事件溯源（Event Sourcing）模式：

将每个技能的状态变更记录为独立事件
事件存储到可靠的消息队列或数据库
通过重放事件重建状态

class SkillEventStore:
    def __init__(self):
        self.events = []

    def append(self, event):
        self.events.append(event)
        # 实际项目中这里应该持久化到数据库

    def get_skill_history(self, skill_id):
        return [e for e in self.events if e.skill_id == skill_id]

import time

class TimedSkill(Skill):
    def execute(self, context):
        start_time = time.time()
        try:
            result = super().execute(context)
            return result
        finally:
            duration = time.time() - start_time
            print(f"Skill {self.skill_id} took {duration:.2f}s")

对共享资源使用锁机制
采用资源池限制并发访问
实现优先级调度

class BatchSkillExecutor(SkillExecutor):
    def submit_batch(self, skills, context):
        """批量提交技能"""
        futures = []
        for skill in skills:
            future = self.submit(skill, context)
            futures.append(future)
        return futures

def execute_with_permission_check(skill, context):
    if not context.user.has_permission(skill.skill_id):
        raise PermissionError("No permission to execute this skill")
    return skill.execute(context)