रचनात्मक पैटर्न

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

# Redis, SQLite, Postgres, आदि हो सकता है।
memory_store: dict[str, dict] = {}

@tool(description="बाद में पुनर्प्राप्ति के लिए महत्वपूर्ण संदर्भ सहेजें")
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="पहले सहेजे गए संदर्भ को पुनर्प्राप्त करें")
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="टैग द्वारा मेमोरी खोजें")
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="सभी सहेजी गई मेमोरी keys")
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:
    """पर्सोना 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="Patient debugging companion")
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:
    # Path validation
    if not any(path.startswith(allowed) for allowed in ALLOWED_PATHS):
        return {"error": f"Access denied: {path}"}

    content = open(path).read()

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

    return {"content": content}

@tool(description="Execute a command (restricted)")
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}"}

    # Execute safely...
    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:]