ai-agent-super-skill

Comprehensive AI agent building skill merging Perplexity Computer's skill creation, webserver, and automation capabilities with Claude Code's agent orchestration, MCP server building, RAG system construction, subagent coordination, parallel agent dispatching, prompt optimization, and execution planning. Covers designing AI agents, building MCP servers, creating RAG pipelines, orchestrating multi-agent systems, optimizing prompts, and deploying AI-powered workflows. Use when building AI agents, creating MCP servers, designing RAG systems, coordinating subagents, optimizing prompts, or architecting any AI-powered automation.

Zur Installation springen

Repository: anavvanzin/iconocracy-corpus
Letzte Quellaktivität: 26. August 2026 um 18:29
Erkannte Sprache von SKILL.md: Englisch
Sterne: 1
Forks: 1

Installationsoptionen

Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.

Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.

Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.

npx skills add https://github.com/anavvanzin/iconocracy-corpus --skill ai-agent-super-skill

Der Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.

Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.

Quelldateien prüfen

Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.

Datei-Explorer

100 Dateien

SKILL.md

Quellanweisungen · Schreibgeschützte Vorschau

Zurück zur Suche Zurück zur Installation

AI Agent Builder Super-Skill

A comprehensive reference for designing, building, and deploying AI agents — from single-tool bots to production multi-agent systems — merging best practices from Claude Code's agent orchestration patterns with Perplexity Computer's deployment infrastructure.

Table of Contents

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { z } from "zod"; // --- Server Initialization --------------------------------------------------- const server = new McpServer({ name: "my-service-mcp", version: "1.0.0", }); // --- Shared API Client -------------------------------------------------------- interface ApiConfig { baseUrl: string; apiKey: string; } class ServiceClient { constructor(private config: ApiConfig) {} async request<T>( endpoint: string, options: RequestInit = {} ): Promise<T> { const url = `${this.config.baseUrl}${endpoint}`; const response = await fetch(url, { ...options, headers: { "Authorization": `Bearer ${this.config.apiKey}`, "Content-Type": "application/json", ...options.headers, }, }); if (!response.ok) { const error = await response.text(); throw new Error( `API error ${response.status}: ${error}. ` + `Suggestion: Check your API key or verify the resource exists.` ); } return response.json() as Promise<T>; } } const client = new ServiceClient({ baseUrl: process.env.SERVICE_BASE_URL ?? "https://api.example.com", apiKey: process.env.SERVICE_API_KEY ?? "", }); // --- Tool: List Items --------------------------------------------------------- server.registerTool( "service_list_items", { description: "List items with optional filtering and pagination.", inputSchema: z.object({ page: z.number().int().min(1).default(1).describe("Page number (1-indexed)"), per_page: z.number().int().min(1).max(100).default(20).describe("Items per page"), filter: z.string().optional().describe("Optional keyword filter"), }), annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, }, }, async (params) => { const query = new URLSearchParams({ page: String(params.page), per_page: String(params.per_page), ...(params.filter ? { q: params.filter } : {}), }); const data = await client.request<{ items: unknown[]; total: number }>( `/items?${query}` ); return { content: [{ type: "text", text: JSON.stringify(data, null, 2), }], structuredContent: data, }; } ); // --- Tool: Get Item ----------------------------------------------------------- server.registerTool( "service_get_item", { description: "Get a single item by ID.", inputSchema: z.object({ id: z.string().describe("Item ID"), }), annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, }, }, async (params) => { const item = await client.request(`/items/${params.id}`); return { content: [{ type: "text", text: JSON.stringify(item, null, 2) }], structuredContent: item, }; } ); // --- Tool: Create Item -------------------------------------------------------- server.registerTool( "service_create_item", { description: "Create a new item. Returns the created item with its assigned ID.", inputSchema: z.object({ name: z.string().min(1).describe("Item name"), description: z.string().optional().describe("Optional item description"), tags: z.array(z.string()).optional().describe("Tag list"), }), annotations: { readOnlyHint: false, destructiveHint: false, idempotentHint: false, }, }, async (params) => { const item = await client.request("/items", { method: "POST", body: JSON.stringify(params), }); return { content: [{ type: "text", text: JSON.stringify(item, null, 2) }], structuredContent: item, }; } ); // --- Transport ---------------------------------------------------------------- const transport = new StdioServerTransport(); await server.connect(transport); console.error("MCP server running on stdio");

#!/usr/bin/env python3 """FastMCP server template for Python-based MCP servers.""" import json import os from typing import Any import httpx from fastmcp import FastMCP from pydantic import BaseModel, Field # --- Server Initialization --------------------------------------------------- mcp = FastMCP("my-service-mcp") # --- API Client --------------------------------------------------------------- BASE_URL = os.environ.get("SERVICE_BASE_URL", "https://api.example.com") API_KEY = os.environ.get("SERVICE_API_KEY", "") async def api_request(endpoint: str, method: str = "GET", body: dict | None = None) -> Any: headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } async with httpx.AsyncClient() as client: response = await client.request( method, f"{BASE_URL}{endpoint}", headers=headers, json=body, ) if not response.is_success: raise ValueError( f"API error {response.status_code}: {response.text}. " "Check your credentials or verify the resource exists." ) return response.json() # --- Input Models ------------------------------------------------------------- class ListParams(BaseModel): page: int = Field(default=1, ge=1, description="Page number (1-indexed)") per_page: int = Field(default=20, ge=1, le=100, description="Items per page") filter: str | None = Field(default=None, description="Keyword filter") class CreateParams(BaseModel): name: str = Field(..., description="Item name") description: str | None = Field(None, description="Optional description") tags: list[str] = Field(default=[], description="Tag list") # --- Tools -------------------------------------------------------------------- @mcp.tool(description="List items with optional filtering and pagination.") async def service_list_items(params: ListParams) -> str: query = f"?page={params.page}&per_page={params.per_page}" if params.filter: query += f"&q={params.filter}" data = await api_request(f"/items{query}") return json.dumps(data, indent=2) @mcp.tool(description="Get a single item by ID.") async def service_get_item(id: str) -> str: """id: Item ID to retrieve""" item = await api_request(f"/items/{id}") return json.dumps(item, indent=2) @mcp.tool(description="Create a new item. Returns the created item with its assigned ID.") async def service_create_item(params: CreateParams) -> str: item = await api_request("/items", method="POST", body=params.model_dump()) return json.dumps(item, indent=2) if __name__ == "__main__": mcp.run()

#!/usr/bin/env python3 """ Production RAG pipeline with chunking, embedding, retrieval, and reranking. """ import hashlib import json from dataclasses import dataclass, field from pathlib import Path from typing import Any # --- Data Models ------------------------------------------------------------- @dataclass class Document: content: str metadata: dict[str, Any] = field(default_factory=dict) doc_id: str = field(default="") def __post_init__(self): if not self.doc_id: self.doc_id = hashlib.md5(self.content.encode()).hexdigest()[:12] @dataclass class Chunk: content: str doc_id: str chunk_index: int metadata: dict[str, Any] = field(default_factory=dict) embedding: list[float] = field(default_factory=list) @dataclass class RetrievedChunk: chunk: Chunk score: float rerank_score: float = 0.0 # --- Chunking Strategies ----------------------------------------------------- class ChunkingStrategy: """Base class — override chunk().""" def chunk(self, doc: Document) -> list[Chunk]: raise NotImplementedError class FixedSizeChunker(ChunkingStrategy): """Fixed token-count chunks with overlap. Good for general text.""" def __init__(self, chunk_size: int = 512, overlap: int = 64): self.chunk_size = chunk_size self.overlap = overlap def chunk(self, doc: Document) -> list[Chunk]: # Simple word-based split (use tiktoken for production token counting) words = doc.content.split() chunks = [] step = self.chunk_size - self.overlap for i in range(0, len(words), step): chunk_words = words[i:i + self.chunk_size] if not chunk_words: break chunks.append(Chunk( content=" ".join(chunk_words), doc_id=doc.doc_id, chunk_index=len(chunks), metadata=doc.metadata, )) return chunks class SentenceChunker(ChunkingStrategy): """Sentence-boundary-aware chunking. Better for structured prose.""" def __init__(self, sentences_per_chunk: int = 5, overlap_sentences: int = 1): self.n = sentences_per_chunk self.ov = overlap_sentences def chunk(self, doc: Document) -> list[Chunk]: import re sentences = re.split(r'(?<=[.!?])\s+', doc.content.strip()) chunks = [] step = self.n - self.ov for i in range(0, len(sentences), step): batch = sentences[i:i + self.n] if not batch: break chunks.append(Chunk( content=" ".join(batch), doc_id=doc.doc_id, chunk_index=len(chunks), metadata=doc.metadata, )) return chunks class RecursiveChunker(ChunkingStrategy): """Hierarchical chunking — preserves section structure. Best for long docs.""" SEPARATORS = ["\n\n", "\n", ". ", " "] def __init__(self, max_chunk_size: int = 800, min_chunk_size: int = 100): self.max_size = max_chunk_size self.min_size = min_chunk_size def chunk(self, doc: Document) -> list[Chunk]: chunks = [] self._split(doc.content, doc.doc_id, doc.metadata, 0, chunks, []) return chunks def _split(self, text, doc_id, metadata, depth, chunks, idx_counter): if len(text.split()) <= self.max_size or depth >= len(self.SEPARATORS): idx_counter.append(None) chunks.append(Chunk( content=text, doc_id=doc_id, chunk_index=len(chunks), metadata=metadata, )) return sep = self.SEPARATORS[depth] parts = text.split(sep) current = "" for part in parts: candidate = (current + sep + part).strip() if current else part if len(candidate.split()) <= self.max_size: current = candidate else: if current and len(current.split()) >= self.min_size: self._split(current, doc_id, metadata, depth + 1, chunks, idx_counter) current = part if current: self._split(current, doc_id, metadata, depth + 1, chunks, idx_counter) # --- Vector Store (using Chroma for local dev) -------------------------------- class VectorStore: """Minimal abstract vector store interface.""" def upsert(self, chunks: list[Chunk]) -> None: raise NotImplementedError def query(self, embedding: list[float], top_k: int = 10, filter_metadata: dict | None = None) -> list[RetrievedChunk]: raise NotImplementedError class ChromaVectorStore(VectorStore): """Local development store using Chroma.""" def __init__(self, collection_name: str, persist_dir: str = "./chroma_db"): import chromadb client = chromadb.PersistentClient(path=persist_dir) self.collection = client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"}, ) def upsert(self, chunks: list[Chunk]) -> None: self.collection.upsert( ids=[f"{c.doc_id}_{c.chunk_index}" for c in chunks], documents=[c.content for c in chunks], embeddings=[c.embedding for c in chunks], metadatas=[c.metadata for c in chunks], ) def query(self, embedding: list[float], top_k: int = 10, filter_metadata: dict | None = None) -> list[RetrievedChunk]: kwargs: dict[str, Any] = { "query_embeddings": [embedding], "n_results": top_k, "include": ["documents", "metadatas", "distances"], } if filter_metadata: kwargs["where"] = filter_metadata results = self.collection.query(**kwargs) retrieved = [] for doc, meta, dist in zip( results["documents"][0], results["metadatas"][0], results["distances"][0], ): retrieved.append(RetrievedChunk( chunk=Chunk(content=doc, doc_id=meta.get("doc_id",""), chunk_index=0, metadata=meta), score=1.0 - dist, # cosine: distance -> similarity )) return retrieved # --- Embedder ---------------------------------------------------------------- class Embedder: """Embed text using OpenAI-compatible API.""" def __init__(self, model: str = "text-embedding-3-small"): import openai self.client = openai.OpenAI() self.model = model def embed(self, texts: list[str]) -> list[list[float]]: response = self.client.embeddings.create(model=self.model, input=texts) return [item.embedding for item in response.data] def embed_one(self, text: str) -> list[float]: return self.embed([text])[0] # --- Reranker ---------------------------------------------------------------- class CrossEncoderReranker: """Rerank retrieved chunks with a cross-encoder for precision improvement.""" def __init__(self, model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"): from sentence_transformers import CrossEncoder self.model = CrossEncoder(model) def rerank(self, query: str, chunks: list[RetrievedChunk], top_k: int = 5) -> list[RetrievedChunk]: pairs = [(query, c.chunk.content) for c in chunks] scores = self.model.predict(pairs) for chunk, score in zip(chunks, scores): chunk.rerank_score = float(score) return sorted(chunks, key=lambda c: c.rerank_score, reverse=True)[:top_k] # --- RAG Pipeline ------------------------------------------------------------ class RAGPipeline: def __init__( self, chunker: ChunkingStrategy, embedder: Embedder, vector_store: VectorStore, reranker: CrossEncoderReranker | None = None, llm_fn: callable = None, retrieval_top_k: int = 10, rerank_top_k: int = 5, ): self.chunker = chunker self.embedder = embedder self.store = vector_store self.reranker = reranker self.llm = llm_fn self.retrieval_k = retrieval_top_k self.rerank_k = rerank_top_k # -- Ingestion ------------------------------------------------------------ def ingest(self, documents: list[Document]) -> int: all_chunks: list[Chunk] = [] for doc in documents: chunks = self.chunker.chunk(doc) texts = [c.content for c in chunks] embeddings = self.embedder.embed(texts) for chunk, emb in zip(chunks, embeddings): chunk.embedding = emb all_chunks.extend(chunks) self.store.upsert(all_chunks) return len(all_chunks) # -- Query ----------------------------------------------------------------- def retrieve( self, query: str, metadata_filter: dict | None = None, use_hyde: bool = False, ) -> list[RetrievedChunk]: # Optional HyDE: generate a hypothetical document to improve retrieval embed_target = query if use_hyde and self.llm: hypothetical = self.llm( f"Write a short passage that would answer this question:\n{query}" ) embed_target = hypothetical q_embedding = self.embedder.embed_one(embed_target) candidates = self.store.query(q_embedding, top_k=self.retrieval_k, filter_metadata=metadata_filter) if self.reranker and candidates: return self.reranker.rerank(query, candidates, top_k=self.rerank_k) return candidates[:self.rerank_k] def answer(self, query: str, metadata_filter: dict | None = None) -> dict: chunks = self.retrieve(query, metadata_filter=metadata_filter) context = "\n\n---\n\n".join( f"[Source {i+1}] {c.chunk.content}" for i, c in enumerate(chunks) ) prompt = f"""Answer the question using ONLY the provided context. If the context doesn't contain the answer, say "I don't have enough information." Context: {context} Question: {query} Answer:""" response = self.llm(prompt) if self.llm else "[No LLM configured]" return { "answer": response, "sources": [ { "content": c.chunk.content[:200] + "...", "score": c.score, "rerank": c.rerank_score, "metadata": c.chunk.metadata, } for c in chunks ], }

from abc import ABC, abstractmethod from dataclasses import dataclass import time import anthropic import openai from tenacity import retry, stop_after_attempt, wait_exponential @dataclass class LLMResponse: content: str model: str input_tokens: int output_tokens: int cost_usd: float latency_ms: float # Cost per 1K tokens (update as pricing changes) PRICING = { "claude-opus-4-5": {"input": 0.015, "output": 0.075}, "claude-haiku-3": {"input": 0.00025, "output": 0.00125}, "gpt-4o": {"input": 0.005, "output": 0.015}, "gpt-4o-mini": {"input": 0.00015, "output": 0.0006}, } class LLMProvider(ABC): @abstractmethod def complete(self, prompt: str, **kwargs) -> LLMResponse: pass def estimate_cost(self, prompt: str, output_tokens_estimate: int = 500) -> float: model = getattr(self, "model", "unknown") pricing = PRICING.get(model, {"input": 0.01, "output": 0.03}) input_tokens = len(prompt.split()) * 1.3 # rough approximation return (input_tokens / 1000 * pricing["input"] + output_tokens_estimate / 1000 * pricing["output"]) class AnthropicProvider(LLMProvider): def __init__(self, model: str = "claude-opus-4-5"): self.client = anthropic.Anthropic() self.model = model @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) def complete(self, prompt: str, max_tokens: int = 2048, **kwargs) -> LLMResponse: start = time.time() response = self.client.messages.create( model=self.model, max_tokens=max_tokens, messages=[{"role": "user", "content": prompt}], **kwargs, ) latency = (time.time() - start) * 1000 pricing = PRICING.get(self.model, {"input": 0.015, "output": 0.075}) cost = (response.usage.input_tokens / 1000 * pricing["input"] + response.usage.output_tokens / 1000 * pricing["output"]) return LLMResponse( content=response.content[0].text, model=self.model, input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens, cost_usd=cost, latency_ms=latency, ) class OpenAIProvider(LLMProvider): def __init__(self, model: str = "gpt-4o-mini"): self.client = openai.OpenAI() self.model = model @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) def complete(self, prompt: str, max_tokens: int = 2048, **kwargs) -> LLMResponse: start = time.time() response = self.client.chat.completions.create( model=self.model, max_tokens=max_tokens, messages=[{"role": "user", "content": prompt}], **kwargs, ) latency = (time.time() - start) * 1000 usage = response.usage pricing = PRICING.get(self.model, {"input": 0.005, "output": 0.015}) cost = (usage.prompt_tokens / 1000 * pricing["input"] + usage.completion_tokens / 1000 * pricing["output"]) return LLMResponse( content=response.choices[0].message.content, model=self.model, input_tokens=usage.prompt_tokens, output_tokens=usage.completion_tokens, cost_usd=cost, latency_ms=latency, ) class FallbackProvider(LLMProvider): """Try primary provider; fall back to secondary on failure.""" def __init__(self, primary: LLMProvider, fallback: LLMProvider): self.primary = primary self.fallback = fallback def complete(self, prompt: str, **kwargs) -> LLMResponse: try: return self.primary.complete(prompt, **kwargs) except Exception as e: print(f"Primary LLM failed ({e}), falling back...") return self.fallback.complete(prompt, **kwargs)

#!/usr/bin/env python3 # cgi-bin/agent_memory.py # Agent memory store: conversations, tool results, learned facts import json import os import sqlite3 import sys from datetime import datetime DB_PATH = "agent_memory.db" def init_db(conn): conn.executescript(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, metadata TEXT DEFAULT '{}' ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN ('user','assistant','tool','system')), content TEXT NOT NULL, tool_name TEXT, tool_input TEXT, tool_result TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES sessions(session_id) ); CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, key TEXT NOT NULL, value TEXT NOT NULL, confidence REAL DEFAULT 1.0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP, UNIQUE(session_id, key) ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id); CREATE INDEX IF NOT EXISTS idx_facts_key ON facts(key); """) conn.commit() conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row init_db(conn) method = os.environ.get("REQUEST_METHOD", "GET") query = os.environ.get("QUERY_STRING", "") path_info = os.environ.get("PATH_INFO", "") def respond(data, status=200): print(f"Status: {status}") print("Content-Type: application/json") print() print(json.dumps(data)) def parse_qs(qs): params = {} for part in qs.split("&"): if "=" in part: k, v = part.split("=", 1) params[k] = v return params # -- Routes -------------------------------------------------------------------- if path_info == "/sessions" and method == "POST": body = json.loads(sys.stdin.read() or "{}") sid = body.get("session_id") or f"sess_{datetime.utcnow().strftime('%Y%m%d_%H%M%S_%f')}" conn.execute("INSERT OR IGNORE INTO sessions (session_id, metadata) VALUES (?,?)", [sid, json.dumps(body.get("metadata", {}))]) conn.commit() respond({"session_id": sid}, 201) elif path_info == "/messages" and method == "POST": body = json.loads(sys.stdin.read()) conn.execute( "INSERT INTO messages (session_id, role, content, tool_name, tool_input, tool_result) " "VALUES (?,?,?,?,?,?)", [body["session_id"], body["role"], body["content"], body.get("tool_name"), body.get("tool_input"), body.get("tool_result")] ) conn.commit() msg_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0] respond({"id": msg_id}, 201) elif path_info == "/messages" and method == "GET": params = parse_qs(query) sid = params.get("session_id", "") limit = int(params.get("limit", 50)) rows = conn.execute( "SELECT * FROM messages WHERE session_id=? ORDER BY created_at LIMIT ?", [sid, limit] ).fetchall() respond([dict(r) for r in rows]) elif path_info == "/facts" and method == "PUT": body = json.loads(sys.stdin.read()) conn.execute( "INSERT OR REPLACE INTO facts (session_id, key, value, confidence) VALUES (?,?,?,?)", [body.get("session_id"), body["key"], json.dumps(body["value"]), body.get("confidence", 1.0)] ) conn.commit() respond({"status": "ok"}) elif path_info == "/facts" and method == "GET": params = parse_qs(query) sid = params.get("session_id", "") rows = conn.execute( "SELECT key, value, confidence FROM facts WHERE session_id=? OR session_id IS NULL", [sid] ).fetchall() respond({r["key"]: {"value": json.loads(r["value"]), "confidence": r["confidence"]} for r in rows}) else: respond({"error": f"Unknown route: {method} {path_info}"}, 400)

#!/usr/bin/env python3 # cgi-bin/webhook_receiver.py # Receives external events and queues them for agent processing import hashlib import hmac import json import os import sqlite3 import sys import time DB_PATH = "webhook_events.db" WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "") conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, event_type TEXT NOT NULL, payload TEXT NOT NULL, processed INTEGER DEFAULT 0, received_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() method = os.environ.get("REQUEST_METHOD", "GET") def verify_signature(body: str, signature: str, secret: str) -> bool: """Verify HMAC-SHA256 webhook signature.""" if not secret: return True # Skip verification if no secret configured expected = "sha256=" + hmac.new( secret.encode(), body.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(signature, expected) if method == "POST": raw_body = sys.stdin.read() sig = os.environ.get("HTTP_X_HUB_SIGNATURE_256", "") if not verify_signature(raw_body, sig, WEBHOOK_SECRET): print("Status: 401") print("Content-Type: application/json") print() print('{"error": "Invalid signature"}') sys.exit(0) body = json.loads(raw_body) conn.execute( "INSERT INTO events (source, event_type, payload) VALUES (?,?,?)", [body.get("source", "unknown"), body.get("type", "unknown"), raw_body] ) conn.commit() print("Status: 202") print("Content-Type: application/json") print() print('{"status": "accepted"}') elif method == "GET": # Dequeue unprocessed events for agent polling rows = conn.execute( "SELECT * FROM events WHERE processed=0 ORDER BY received_at LIMIT 50" ).fetchall() events = [ {"id": r[0], "source": r[1], "event_type": r[2], "payload": json.loads(r[3]), "received_at": r[5]} for r in rows ] print("Content-Type: application/json") print() print(json.dumps(events))

#!/usr/bin/env python3 # cgi-bin/message_bus.py # Simple pub/sub message bus for multi-agent coordination import json import os import sqlite3 import sys import uuid DB_PATH = "message_bus.db" conn = sqlite3.connect(DB_PATH) conn.executescript(""" CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, from_agent TEXT NOT NULL, to_agent TEXT, -- NULL = broadcast topic TEXT NOT NULL, payload TEXT NOT NULL, ack INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS subscriptions ( agent_id TEXT NOT NULL, topic TEXT NOT NULL, PRIMARY KEY (agent_id, topic) ); """) conn.commit() method = os.environ.get("REQUEST_METHOD", "GET") path_info = os.environ.get("PATH_INFO", "") query = os.environ.get("QUERY_STRING", "") def respond(data, status=200): print(f"Status: {status}") print("Content-Type: application/json") print() print(json.dumps(data)) if path_info == "/publish" and method == "POST": body = json.loads(sys.stdin.read()) msg_id = str(uuid.uuid4()) conn.execute( "INSERT INTO messages (id, from_agent, to_agent, topic, payload) VALUES (?,?,?,?,?)", [msg_id, body["from_agent"], body.get("to_agent"), body["topic"], json.dumps(body["payload"])] ) conn.commit() respond({"message_id": msg_id}, 201) elif path_info == "/subscribe" and method == "POST": body = json.loads(sys.stdin.read()) conn.execute( "INSERT OR IGNORE INTO subscriptions VALUES (?,?)", [body["agent_id"], body["topic"]] ) conn.commit() respond({"status": "subscribed"}) elif path_info == "/poll" and method == "GET": params = dict(p.split("=") for p in query.split("&") if "=" in p) agent_id = params.get("agent_id", "") # Get messages for this agent (direct + subscribed topics) rows = conn.execute(""" SELECT m.* FROM messages m LEFT JOIN subscriptions s ON s.agent_id=? AND s.topic=m.topic WHERE m.ack=0 AND (m.to_agent=? OR (m.to_agent IS NULL AND s.agent_id IS NOT NULL)) ORDER BY m.created_at LIMIT 20 """, [agent_id, agent_id]).fetchall() messages = [{"id": r[0], "from": r[1], "topic": r[3], "payload": json.loads(r[4])} for r in rows] # Mark as acked if messages: ids = [m["id"] for m in messages] conn.execute(f"UPDATE messages SET ack=1 WHERE id IN ({','.join('?'*len(ids))})", ids) conn.commit() respond(messages)

// Agent backend client — use __CGI_BIN__ placeholder in production const CGI_BIN = "__CGI_BIN__"; class AgentMemoryClient { async createSession(metadata = {}) { const res = await fetch(`${CGI_BIN}/agent_memory.py/sessions`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ metadata }), }); return res.json(); } async addMessage(sessionId, role, content, toolData = {}) { const res = await fetch(`${CGI_BIN}/agent_memory.py/messages`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: sessionId, role, content, ...toolData }), }); return res.json(); } async getHistory(sessionId, limit = 50) { const res = await fetch( `${CGI_BIN}/agent_memory.py/messages?session_id=${sessionId}&limit=${limit}` ); return res.json(); } async setFact(sessionId, key, value, confidence = 1.0) { const res = await fetch(`${CGI_BIN}/agent_memory.py/facts`, { method: "PUT", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ session_id: sessionId, key, value, confidence }), }); return res.json(); } async getFacts(sessionId) { const res = await fetch( `${CGI_BIN}/agent_memory.py/facts?session_id=${sessionId}` ); return res.json(); } } class MessageBusClient { async publish(fromAgent, topic, payload, toAgent = null) { const res = await fetch(`${CGI_BIN}/message_bus.py/publish`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ from_agent: fromAgent, topic, payload, to_agent: toAgent }), }); return res.json(); } async subscribe(agentId, topic) { const res = await fetch(`${CGI_BIN}/message_bus.py/subscribe`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ agent_id: agentId, topic }), }); return res.json(); } async poll(agentId) { const res = await fetch(`${CGI_BIN}/message_bus.py/poll?agent_id=${agentId}`); return res.json(); } }

Capability	Source Skill(s)	Coverage	Gaps Filled Here
Agent architecture (ReAct, Plan-Execute)	senior-prompt-engineer	Partial — workflow diagrams only	Full pattern library with code
Multi-agent orchestration	subagent-driven-development, dispatching-parallel-agents	Strong process, no code	Integration patterns, conflict detection
MCP server building	mcp-builder	Full (4-phase process)	Perplexity-compatible CGI deployment
RAG pipeline construction	senior-ml-engineer, senior-prompt-engineer	Chunking + DB selection tables	End-to-end pipeline code
Prompt engineering	senior-prompt-engineer	Pattern reference table	Advanced chain-of-thought + meta-prompting
Subagent task dispatch	subagent-driven-development	Process diagrams	Template prompts with full context injection
Parallel agent dispatch	dispatching-parallel-agents	Decision tree + examples	Conflict detection, state isolation
Plan execution with checkpoints	executing-plans	Step-by-step process	Batch sizing, rollback strategies
MLOps / model deployment	senior-ml-engineer	Docker + k8s templates	Agent-specific serving patterns
Backend webhooks/SQLite	webserver	Full CGI-bin reference	Agent memory persistence layer
Skill packaging (SKILL.md)	create-skill (Perplexity)	YAML frontmatter format	Validation pipeline, versioning
Deployment & observability	website-building (Perplexity)	UI deployment only	Agent health checks, trace logging
Perplexity 400+ integrations	Perplexity Computer native	Available but undocumented	Integration mapping for agent use
Scheduled monitoring	Perplexity Computer native	Not in any skill	Agent heartbeat and drift triggers

Goal	Pattern	Reason
Open-ended research	ReAct	Flexible, self-correcting, handles unknown paths
Multi-step report generation	Plan-Execute	Predictable, auditable, checkpointable
Code writing / debugging	Reflexion	Self-critique loop improves quality over iterations
API integration / tool calling	Tool-Use	Native LLM feature, lower latency, less prompt engineering
Customer support bot	ReAct + Tool-Use	Hybrid: structured tools with flexible reasoning
Batch data processing	Plan-Execute with parallel dispatch	Speed via parallelism, structured output
Creative tasks (writing, design)	Reflexion	Quality improves with each self-critique cycle

Database	Hosting	Scale	Latency	Best For
Pinecone	Managed cloud	High	Low	Production, zero-ops
Qdrant	Self-hosted / cloud	High	Very Low	Performance-critical
Weaviate	Both	High	Low	Hybrid (keyword + vector)
Chroma	Self-hosted	Medium	Low	Local dev / prototyping
pgvector	Self-hosted (Postgres)	Medium	Medium	Existing Postgres stacks
Redis VSS	Both	Medium	Very Low	Real-time / cache-adjacent
Milvus	Self-hosted / cloud	Very High	Low	Enterprise scale

Strategy	Chunk Size	Overlap	Best For
Fixed-size	500-1000 tokens	50-100 tokens	General text, unknown structure
Sentence	3-5 sentences	1 sentence	News articles, documentation
Semantic	Variable	Meaning-based	Research papers, books
Recursive	Hierarchical	Parent-child	Long documents with headers

Metric	Definition	Target
Context Relevance	% of retrieved chunks relevant to query	> 0.80
Answer Faithfulness	% of answer grounded in context	> 0.90
Retrieval Precision@5	Relevant chunks in top 5 / 5	> 0.70
Context Coverage	% of questions with >=1 relevant chunk in top-5	> 0.85
End-to-end Accuracy	Correct answers / total questions	> 0.80

name	ai-agent-super-skill
description	Comprehensive AI agent building skill merging Perplexity Computer's skill creation, webserver, and automation capabilities with Claude Code's agent orchestration, MCP server building, RAG system construction, subagent coordination, parallel agent dispatching, prompt optimization, and execution planning. Covers designing AI agents, building MCP servers, creating RAG pipelines, orchestrating multi-agent systems, optimizing prompts, and deploying AI-powered workflows. Use when building AI agents, creating MCP servers, designing RAG systems, coordinating subagents, optimizing prompts, or architecting any AI-powered automation.
license	MIT
metadata	{"author":"get-zeked","version":"1.0"}

Pattern	When to Use	Token Cost	Quality Gain
Zero-shot	Simple, well-defined tasks	Lowest	Baseline
Few-shot (3-5 examples)	Complex tasks, consistent format needed	Medium	High
Chain-of-Thought (CoT)	Reasoning, math, multi-step logic	Medium	High
Role Prompting	Domain expertise, specific perspective	Low	Medium
Structured Output	Need parseable JSON/XML	Low	High (reliability)
Tree-of-Thought	Complex problem solving, backtracking	High	Very High
Meta-prompting	Generating/optimizing other prompts	High	Very High
Self-consistency	High-stakes decisions (majority vote)	Very High	High

Strategy	Latency	Throughput	Cost	Use Case
FastAPI + Uvicorn	Low	Medium	Low	REST agent APIs, single-model
Ray Serve	Medium	Very High	Medium	Multi-model pipelines, scaling
Triton Inference	Very Low	Very High	Medium	GPU batch inference
Serverless (Lambda/Cloud Run)	Cold-start medium	Auto-scale	Pay-per-use	Bursty agent tasks
Streaming (SSE/WebSocket)	Apparent Low	Medium	Low	Conversational agents

ai-agent-super-skill

Quellinformationen

Installationsoptionen

Quelldateien prüfen

Mehr aus diesem Repository

AI Agent Builder Super-Skill

Table of Contents

Mehr aus diesem Repository

AI Agent Builder Super-Skill

Table of Contents

1. Gap Analysis Table

2. Agent Architecture & Design Patterns

2.1 Core Agent Loop

2.2 Architecture Patterns

Pattern A: ReAct (Reason + Act)

Pattern B: Plan-and-Execute

Pattern C: Reflexion

Pattern D: Tool-Use Agent (Function Calling)

2.3 Architecture Selection Guide

2.4 Multi-Agent System Topologies

3. MCP Server Development

3.1 Four-Phase MCP Build Process

3.2 TypeScript MCP Server Template

3.3 Python FastMCP Server Template

3.4 MCP Tool Design Checklist

3.5 MCP Error Message Patterns

4. RAG System Construction

4.1 Complete RAG Pipeline

4.2 Full Python RAG Implementation

4.3 Vector Database Selection

4.4 Chunking Strategy Selection

4.5 RAG Evaluation Metrics

5. Subagent Coordination

5.1 Subagent-Driven Development

5.2 Implementer Subagent Prompt Template

5.3 Spec Reviewer Prompt Template

5.4 Code Quality Reviewer Prompt Template

5.5 Red Flags in Subagent Coordination

6. Execution Planning & Verification

6.1 Parallel Agent Dispatch Pattern

6.2 Focused Agent Task Prompt Structure

6.3 Batch Execution with Checkpoints

6.4 Verification Workflow

7. Prompt Engineering & Optimization

7.1 Core Prompt Patterns

7.2 Chain-of-Thought Implementation

7.3 Structured Output Design

7.4 Prompt Optimization Workflow

7.5 Meta-Prompting (Prompt Generation)

7.6 Few-Shot Example Design

8. ML Integration for Agents

8.1 LLM Provider Abstraction Layer

8.2 Model Deployment for Agent Serving

8.3 Model Monitoring for Agents

8.4 Serving Strategy Selection

9. Skill & Capability Creation

9.1 SKILL.md Format Specification

9.2 Validation Pipeline

9.3 Skill Quality Checklist

9.4 Skill Packaging for Distribution

9.5 Skill Trigger Design

10. Backend Infrastructure for Agents

10.1 Agent Memory Persistence with SQLite (CGI-bin)

10.2 Webhook Receiver for Agent Triggers

10.3 Agent-to-Agent Communication via Message Bus

10.4 JavaScript Client for Agent Backend

11. Agent Deployment & Monitoring

11.1 Deployment Checklist

11.2 Structured Agent Logging

11.3 Agent Health Dashboard Schema