创意模式

from dataclasses import dataclass, field
from dedalus_mcp import MCPServer, tool

@dataclass
class ThinkingState:
    thoughts: list[dict] = field(default_factory=list)
    branches: dict[str, list[dict]] = field(default_factory=dict)

state = ThinkingState()

@tool(description="""逐步推理,支持修订。

使用场景:
- 分解复杂问题
- 需要调整方向的规划
- 初始范围不明确的分析

您可以修订之前的想法、分支到替代方案或超出最初的估计继续扩展。""")
def think(
    thought: str,
    thought_number: int,
    total_thoughts: int,
    next_thought_needed: bool,
    is_revision: bool = False,
    revises_thought: int | None = None,
    branch_id: str | None = None,
) -> dict:
    entry = {
        "number": thought_number,
        "thought": thought,
        "is_revision": is_revision,
        "revises": revises_thought,
    }

    if branch_id:
        state.branches.setdefault(branch_id, []).append(entry)
    else:
        state.thoughts.append(entry)

    return {
        "thought_number": thought_number,
        "total_thoughts": total_thoughts,
        "next_thought_needed": next_thought_needed,
        "history_length": len(state.thoughts),
        "branches": list(state.branches.keys()),
    }

server = MCPServer("reasoning")
server.collect(think)

from dedalus_mcp import MCPServer, tool, resource
import json

# Could be Redis, SQLite, Postgres, etc.
memory_store: dict[str, dict] = {}

@tool(description="Save important context for later retrieval")
def remember(key: str, content: str, tags: list[str] | None = None) -> dict:
    memory_store[key] = {
        "content": content,
        "tags": tags or [],
        "saved_at": datetime.now().isoformat(),
    }
    return {"saved": key}

@tool(description="Retrieve previously saved context")
def recall(key: str) -> dict:
    if key not in memory_store:
        return {"error": f"No memory for key: {key}"}
    return memory_store[key]

@tool(description="Find memories by tag")
def search_memories(tag: str) -> list[dict]:
    return [
        {"key": k, **v}
        for k, v in memory_store.items()
        if tag in v.get("tags", [])
    ]

@resource(uri="memory://index", description="所有已保存的记忆键")
def memory_index() -> dict:
    return {
        "keys": list(memory_store.keys()),
        "count": len(memory_store),
    }

server = MCPServer("memory")
server.collect(remember, recall, search_memories, memory_index)

from dedalus_mcp import MCPServer, resource
import asyncio

metrics = {"cpu": 0.0, "memory": 0.0, "requests": 0}

@resource(uri="system://metrics", description="实时系统指标")
def get_metrics() -> dict:
    return {"timestamp": datetime.now().isoformat(), **metrics}

server = MCPServer("monitoring")
server.collect(get_metrics)

async def update_metrics():
    while True:
        metrics["cpu"] = get_cpu_usage()
        metrics["memory"] = get_memory_usage()
        metrics["requests"] = get_request_count()
        await server.notify_resource_updated("system://metrics")
        await asyncio.sleep(5)

from dataclasses import dataclass
from dedalus_mcp import MCPServer, prompt, Message

@dataclass(frozen=True)
class PersonaArgs:
    """Arguments for persona prompts."""
    context: str | None = None
    verbosity: str = "normal"
    focus_area: str | None = None

@prompt("persona/architect", description="Senior software architect")
def architect_persona(args: PersonaArgs) -> list[Message]:
    context = f" Current context: {args.context}" if args.context else ""
    return [
        Message(role="assistant", content=f"""You are a senior software architect with 20 years of experience.
You think in systems, not features. You ask clarifying questions before proposing solutions.
You consider maintainability, scalability, and team dynamics.{context}"""),
    ]

@prompt("persona/reviewer", description="Strict code reviewer")
def reviewer_persona(args: PersonaArgs) -> list[Message]:
    focus = f" Focus especially on {args.focus_area}." if args.focus_area else ""
    return [
        Message(role="assistant", content=f"""You are a meticulous code reviewer.
You catch bugs others miss. You insist on tests.
You're constructive but don't let things slide.{focus}"""),
    ]

@prompt("persona/rubber-duck", description="耐心的调试伙伴")
def rubber_duck_persona(args: PersonaArgs) -> list[Message]:
    return [
        Message(role="assistant", content="""You help by asking questions, not giving answers.
When someone explains their problem, ask what they've tried.
Help them think through it systematically."""),
    ]

server = MCPServer("personas")
server.collect(architect_persona, reviewer_persona, rubber_duck_persona)

from dedalus_mcp import MCPServer, tool

ALLOWED_PATHS = ["/app/data", "/app/config"]
BLOCKED_PATTERNS = ["password", "secret", "api_key"]

@tool(description="读取文件(含安全检查)")
def safe_read(path: str) -> dict:
    # 路径验证
    if not any(path.startswith(allowed) for allowed in ALLOWED_PATHS):
        return {"error": f"Access denied: {path}"}

    content = open(path).read()

    # 内容过滤
    for pattern in BLOCKED_PATTERNS:
        if pattern in content.lower():
            return {"error": "Content contains sensitive data"}

    return {"content": content}

@tool(description="执行命令(受限)")
def safe_exec(command: str) -> dict:
    allowed = ["ls", "cat", "grep", "find"]
    cmd = command.split()[0]

    if cmd not in allowed:
        return {"error": f"Command not allowed: {cmd}"}

    # 安全执行...
    return {"output": "..."}

from dedalus_mcp import MCPServer, tool, resource
from enum import Enum

class WorkflowStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

workflows: dict[str, dict] = {}

@tool(description="启动新工作流")
def start_workflow(workflow_id: str, steps: list[str]) -> dict:
    workflows[workflow_id] = {
        "status": WorkflowStatus.RUNNING.value,
        "steps": steps,
        "current_step": 0,
        "results": [],
    }
    return {"workflow_id": workflow_id, "status": "started"}

@tool(description="完成当前步骤并继续")
def complete_step(workflow_id: str, result: str) -> dict:
    wf = workflows.get(workflow_id)
    if not wf:
        return {"error": "Workflow not found"}

    wf["results"].append(result)
    wf["current_step"] += 1

    if wf["current_step"] >= len(wf["steps"]):
        wf["status"] = WorkflowStatus.COMPLETED.value

    return {
        "step_completed": wf["current_step"],
        "next_step": wf["steps"][wf["current_step"]] if wf["current_step"] < len(wf["steps"]) else None,
        "status": wf["status"],
    }

@resource(uri="workflows://active", description="所有活跃的工作流")
def active_workflows() -> dict:
    return {
        wid: wf for wid, wf in workflows.items()
        if wf["status"] == WorkflowStatus.RUNNING.value
    }

from dedalus_mcp import MCPServer, tool, resource
from datetime import datetime

audit_log: list[dict] = []

def log_action(action: str, details: dict):
    audit_log.append({
        "timestamp": datetime.now().isoformat(),
        "action": action,
        **details,
    })

@tool(description="执行敏感操作")
def sensitive_operation(operation: str, target: str) -> dict:
    log_action("sensitive_operation", {"operation": operation, "target": target})
    # ... 执行具体操作 ...
    return {"status": "completed"}

@resource(uri="audit://log", description="完整审计日志")
def get_audit_log() -> list[dict]:
    return audit_log

@resource(uri="audit://recent", description="最近 10 次操作")
def recent_actions() -> list[dict]:
    return audit_log[-10:]