Spaces:
Sleeping
Sleeping
| """ | |
| Autonomous AI Agent for Hugging Face Spaces | |
| Gradio 6.14.0 + OpenRouter Free Models + ClawHub Skills + MCP Protocol | |
| Supports: | |
| - ReAct agent loop with tool calling | |
| - Free OpenRouter models with automatic fallback | |
| - ClawHub skill installation via Convex API | |
| - MCP Server: this Space is discoverable and callable by any MCP client | |
| (Claude Desktop, ChatGPT, Cursor, VS Code, other agents) | |
| - MCP Client: can connect to other MCP-enabled Gradio Spaces and use their tools | |
| - A2A (Agent-to-Agent) networking: public registry, auto-discovery, task delegation | |
| - SkillClaw: collective skill evolution via public skill cloud | |
| - HF Spaces Scanner: find other agents on HuggingFace | |
| - Moltbook: social network for AI agents (requires API key) | |
| - artinet.io: agent discovery service (best-effort, often unreachable) | |
| Key HF Integration: | |
| - mcp_server=True in demo.launch() exposes tools via MCP protocol | |
| - MCP endpoint: /gradio_api/mcp/ (auto-generated by Gradio) | |
| - Any MCP-compatible client can discover and use this agent's tools | |
| - This agent can also call tools from other MCP-enabled Spaces | |
| """ | |
| import os | |
| import io | |
| import json | |
| import re | |
| import zipfile | |
| import time | |
| import hashlib | |
| import requests | |
| import gradio as gr | |
| from starlette.responses import JSONResponse | |
| from starlette.routing import Route | |
| # ─── Configuration ────────────────────────────────────────────── | |
| OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "") | |
| FREE_MODELS = [ | |
| "qwen/qwen3-coder:free", | |
| "nvidia/nemotron-3-super-120b-a12b:free", | |
| "google/gemma-4-31b-it:free", | |
| "qwen/qwen3-next-80b-a3b-instruct:free", | |
| "openai/gpt-oss-120b:free", | |
| "z-ai/glm-4.5-air:free", | |
| "meta-llama/llama-3.3-70b-instruct:free", | |
| "nvidia/nemotron-3-nano-30b-a3b:free", | |
| ] | |
| API_URL = "https://openrouter.ai/api/v1/chat/completions" | |
| MAX_ITERATIONS = 10 | |
| MAX_SKILL_PROMPT_CHARS = 6000 | |
| DATA_DIR = "/data" | |
| SKILLS_DIR = os.path.join(DATA_DIR, "skills") | |
| PEERS_FILE = os.path.join(DATA_DIR, "peers.json") | |
| MCP_SERVERS_FILE = os.path.join(DATA_DIR, "mcp_servers.json") | |
| AGENT_NAME = os.environ.get("AGENT_NAME", "autonomous-agent") | |
| AGENT_DESCRIPTION = os.environ.get( | |
| "AGENT_DESCRIPTION", | |
| "An autonomous AI agent with web search, code execution, file management, " | |
| "ClawHub skill installation, MCP protocol support, A2A open peer networking, " | |
| "and HF Spaces peer discovery capabilities.", | |
| ) | |
| A2A_VERSION = "0.3" | |
| # ─── A2A Public Network Registry (jsonbin.io) ────────────────── | |
| A2A_REGISTRY_KEY = os.environ.get("A2A_REGISTRY_KEY", "") | |
| A2A_REGISTRY_BIN = os.environ.get("A2A_REGISTRY_BIN", "") | |
| A2A_REGISTRY_API = "https://api.jsonbin.io/v3" | |
| A2A_SKILLCLOUD_BIN = os.environ.get("A2A_SKILLCLOUD_BIN", "") | |
| # ─── Moltbook (Agent Social Network) ─────────────────────────── | |
| MOLTBOOK_API_KEY = os.environ.get("MOLTBOOK_API_KEY", "") | |
| MOLTBOOK_API = "https://www.moltbook.com/api/v1" | |
| # ─── artinet.io (Agent Discovery) ────────────────────────────── | |
| ARTINET_API = "https://api.artinet.io" | |
| os.makedirs(SKILLS_DIR, exist_ok=True) | |
| # ─── Load saved A2A config from /data (survives restarts) ────── | |
| def _load_a2a_config(): | |
| """Load saved bin IDs from /data/a2a_config.json.""" | |
| global A2A_REGISTRY_BIN, A2A_SKILLCLOUD_BIN | |
| config_path = os.path.join(DATA_DIR, "a2a_config.json") | |
| if os.path.exists(config_path): | |
| try: | |
| with open(config_path, "r") as f: | |
| cfg = json.load(f) | |
| if not A2A_REGISTRY_BIN and cfg.get("registry_bin"): | |
| A2A_REGISTRY_BIN = cfg["registry_bin"] | |
| if not A2A_SKILLCLOUD_BIN and cfg.get("skillcloud_bin"): | |
| A2A_SKILLCLOUD_BIN = cfg["skillcloud_bin"] | |
| except Exception: | |
| pass | |
| _load_a2a_config() | |
| def _save_a2a_config(**kwargs): | |
| """Save A2A config (bin IDs) to /data/a2a_config.json.""" | |
| config_path = os.path.join(DATA_DIR, "a2a_config.json") | |
| cfg = {} | |
| if os.path.exists(config_path): | |
| try: | |
| with open(config_path, "r") as f: | |
| cfg = json.load(f) | |
| except Exception: | |
| pass | |
| cfg.update(kwargs) | |
| with open(config_path, "w") as f: | |
| json.dump(cfg, f, indent=2) | |
| # ─── LLM Client ──────────────────────────────────────────────── | |
| def call_llm(messages, tools=None): | |
| """Call OpenRouter API with free model fallback.""" | |
| headers = { | |
| "Authorization": f"Bearer {OPENROUTER_API_KEY}", | |
| "Content-Type": "application/json", | |
| "HTTP-Referer": "https://huggingface.co", | |
| "X-Title": AGENT_NAME, | |
| } | |
| payload = { | |
| "model": "", | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 4096, | |
| } | |
| if tools: | |
| payload["tools"] = tools | |
| payload["tool_choice"] = "auto" | |
| last_error = None | |
| for model in FREE_MODELS: | |
| payload["model"] = model | |
| try: | |
| resp = requests.post(API_URL, headers=headers, json=payload, timeout=120) | |
| if resp.status_code == 402: | |
| last_error = f"{model}: 402 Payment Required" | |
| continue | |
| if resp.status_code == 429: | |
| last_error = f"{model}: 429 Rate Limited" | |
| continue | |
| if resp.status_code == 404: | |
| last_error = f"{model}: 404 Not Found" | |
| continue | |
| resp.raise_for_status() | |
| result = resp.json() | |
| result["_model_used"] = model | |
| return result | |
| except requests.exceptions.HTTPError as e: | |
| last_error = f"{model}: {e}" | |
| continue | |
| except Exception as e: | |
| last_error = f"{model}: {e}" | |
| continue | |
| return {"error": f"All free models failed. Last: {last_error}"} | |
| # ─── Tool Definitions ────────────────────────────────────────── | |
| TOOL_DEFINITIONS = [ | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "web_search", | |
| "description": "Search the web using DuckDuckGo.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"query": {"type": "string", "description": "Search query"}}, | |
| "required": ["query"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "read_webpage", | |
| "description": "Fetch and extract text content from a URL.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"url": {"type": "string", "description": "URL to fetch"}}, | |
| "required": ["url"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "execute_python", | |
| "description": "Execute Python code and return stdout.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"code": {"type": "string", "description": "Python code"}}, | |
| "required": ["code"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "write_file", | |
| "description": "Write content to a file in the persistent /data directory.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "path": {"type": "string", "description": "File path relative to /data"}, | |
| "content": {"type": "string", "description": "Content to write"}, | |
| }, | |
| "required": ["path", "content"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "read_file", | |
| "description": "Read content from a file in /data.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": {"path": {"type": "string", "description": "File path relative to /data"}}, | |
| "required": ["path"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "list_files", | |
| "description": "List files in /data directory.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "subdir": {"type": "string", "description": "Subdirectory relative to /data"} | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "install_skill", | |
| "description": "Install a skill from ClawHub (e.g. 'darkd/integral-hermeneutics').", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "skill_id": {"type": "string", "description": "Skill ID in format 'author/skill-name'"} | |
| }, | |
| "required": ["skill_id"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "list_skills", | |
| "description": "List all installed skills and their active/inactive status.", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "toggle_skill", | |
| "description": "Activate or deactivate an installed skill.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "name": {"type": "string", "description": "Skill name"}, | |
| "active": {"type": "boolean", "description": "True to activate, False to deactivate"}, | |
| }, | |
| "required": ["name", "active"], | |
| }, | |
| }, | |
| }, | |
| # ─── MCP Protocol Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "call_mcp_tool", | |
| "description": "Call a tool on a remote MCP-enabled Gradio Space. Use this to access capabilities of other HF Spaces (image generation, transcription, translation, etc.).", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "space_url": { | |
| "type": "string", | |
| "description": "The Space URL (e.g. 'https://username-spacename.hf.space')", | |
| }, | |
| "tool_name": { | |
| "type": "string", | |
| "description": "Name of the MCP tool to call (use list_mcp_tools to discover available tools)", | |
| }, | |
| "arguments": { | |
| "type": "object", | |
| "description": "Arguments to pass to the tool", | |
| }, | |
| }, | |
| "required": ["space_url", "tool_name"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "list_mcp_tools", | |
| "description": "List available MCP tools on a remote Gradio Space. Discover what tools another Space exposes via the MCP protocol.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "space_url": { | |
| "type": "string", | |
| "description": "The Space URL (e.g. 'https://username-spacename.hf.space')", | |
| }, | |
| }, | |
| "required": ["space_url"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "browse_mcp_spaces", | |
| "description": "Browse HuggingFace Spaces that support the MCP protocol. These Spaces expose tools that any MCP client (including this agent) can call. This is the best way to find useful tools from the HF community.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search query (e.g. 'image generation', 'transcription', 'translation'). Leave empty for trending.", | |
| }, | |
| "limit": { | |
| "type": "integer", | |
| "description": "Max spaces to return (default 15, max 30)", | |
| }, | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| # ─── A2A Open Network Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "register_on_network", | |
| "description": "Register this agent on the public A2A network so ANY other agent can discover and delegate tasks to it.", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "discover_peers", | |
| "description": "Search the public A2A network for agents matching a skill or capability.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search query -- skill name, capability, or description keywords", | |
| } | |
| }, | |
| "required": ["query"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "browse_network", | |
| "description": "List all agents registered on the public A2A network.", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "add_peer", | |
| "description": "Manually register a peer agent by its URL. Fetches its A2A Agent Card to discover capabilities.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "url": { | |
| "type": "string", | |
| "description": "Peer agent URL (e.g. 'https://username-agentname.hf.space')", | |
| } | |
| }, | |
| "required": ["url"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "list_peers", | |
| "description": "List all known peer agents (both from network discovery and manual add).", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "delegate_to_peer", | |
| "description": "Delegate a task to a peer agent. Auto-selects best peer by affinity if no URL given.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "task": {"type": "string", "description": "The task to delegate"}, | |
| "peer_url": { | |
| "type": "string", | |
| "description": "Peer agent URL (optional -- auto-selects best peer if omitted)", | |
| }, | |
| }, | |
| "required": ["task"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "share_skill_with_peer", | |
| "description": "Tell a peer agent to install a ClawHub skill.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "peer_url": {"type": "string", "description": "Peer agent URL"}, | |
| "skill_id": { | |
| "type": "string", | |
| "description": "ClawHub skill ID to share (e.g. 'darkd/integral-hermeneutics')", | |
| }, | |
| }, | |
| "required": ["peer_url", "skill_id"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "query_peer", | |
| "description": "Ask a peer agent a question and get its response.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "peer_url": {"type": "string", "description": "Peer agent URL"}, | |
| "query": {"type": "string", "description": "Question to ask"}, | |
| }, | |
| "required": ["peer_url", "query"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "evolve_skill", | |
| "description": "Create an improved version of an installed skill based on usage notes.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "skill_name": {"type": "string", "description": "Name of the skill to evolve"}, | |
| "improvements": { | |
| "type": "string", | |
| "description": "Description of improvements or adaptations to make", | |
| }, | |
| }, | |
| "required": ["skill_name", "improvements"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "push_skill_to_network", | |
| "description": "Share an evolved skill with the public SkillClaw cloud.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "skill_name": { | |
| "type": "string", | |
| "description": "Name of the locally evolved skill to share", | |
| } | |
| }, | |
| "required": ["skill_name"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "pull_skills_from_network", | |
| "description": "Browse the public SkillClaw cloud for skills shared by other agents.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search query -- skill name, keyword, or description", | |
| } | |
| }, | |
| "required": ["query"], | |
| }, | |
| }, | |
| }, | |
| # ─── Moltbook Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "moltbook_register", | |
| "description": "Register this agent on Moltbook -- the front page of the agent internet. Requires MOLTBOOK_API_KEY.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "username": {"type": "string", "description": "Username for this agent on Moltbook"}, | |
| "bio": {"type": "string", "description": "Short bio describing this agent's capabilities"}, | |
| }, | |
| "required": ["username", "bio"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "moltbook_post", | |
| "description": "Create a post on Moltbook in a submolt. Requires MOLTBOOK_API_KEY.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "title": {"type": "string", "description": "Post title"}, | |
| "content": {"type": "string", "description": "Post content (markdown supported)"}, | |
| "submolt": {"type": "string", "description": "Submolt to post in (e.g. 's/introductions', 's/showcase')"}, | |
| }, | |
| "required": ["title", "content", "submolt"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "moltbook_feed", | |
| "description": "Browse the Moltbook feed or a specific submolt to discover other AI agents. Requires MOLTBOOK_API_KEY.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "submolt": { | |
| "type": "string", | |
| "description": "Submolt name. Leave empty for the main feed.", | |
| }, | |
| "limit": { | |
| "type": "integer", | |
| "description": "Number of posts to return (default 10, max 25)", | |
| }, | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| # ─── artinet.io Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "artinet_register", | |
| "description": "Register this agent on artinet.io -- a dedicated agent discovery service. No API key needed.", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "artinet_search", | |
| "description": "Search for agents on artinet.io by capability or keyword.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search query -- capability, keyword, or agent type", | |
| } | |
| }, | |
| "required": ["query"], | |
| }, | |
| }, | |
| }, | |
| # ─── HF Spaces Scanner Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "scan_hf_spaces", | |
| "description": "Scan HuggingFace Hub for other AI agents. NO API KEY NEEDED! The most reliable peer discovery method.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "Search query (e.g. 'agent', 'chatbot', 'assistant'). Leave empty for trending.", | |
| }, | |
| "limit": { | |
| "type": "integer", | |
| "description": "Max spaces to return (default 15, max 30)", | |
| }, | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "scan_alive_peers", | |
| "description": "Ping all known peer URLs to check which ones are alive and responding.", | |
| "parameters": {"type": "object", "properties": {}, "required": []}, | |
| }, | |
| }, | |
| # ─── Skill Sharing Tools ──── | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "offer_skill_to_space", | |
| "description": "Offer one of our installed skills to another Gradio Space via MCP or chat API. This is the primary method for sharing skills with other agents. Tries multiple approaches: MCP tool call, Gradio chat API, and direct HTTP skill reference.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "space_url": { | |
| "type": "string", | |
| "description": "URL of the target Gradio Space (e.g. 'https://username-spacename.hf.space')", | |
| }, | |
| "skill_name": { | |
| "type": "string", | |
| "description": "Name of our installed skill to share (use list_skills to see available)", | |
| }, | |
| }, | |
| "required": ["space_url", "skill_name"], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "find_collaborators", | |
| "description": "Find other AI agents on HuggingFace that might be receptive to skill sharing. Searches for MCP-enabled spaces, chatbot agents, and agents with tool-accepting endpoints. Returns a ranked list with tips on how to approach each one.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "query": { | |
| "type": "string", | |
| "description": "What kind of agent to look for (e.g. 'hermeneutics', 'divination', 'analysis'). Leave empty for general search.", | |
| }, | |
| "limit": { | |
| "type": "integer", | |
| "description": "Max results (default 10, max 20)", | |
| }, | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| { | |
| "type": "function", | |
| "function": { | |
| "name": "get_shared_skills", | |
| "description": "Get a list of skills this agent can share with others, including full skill data suitable for installation. Other agents can call this to discover and fetch our skills.", | |
| "parameters": { | |
| "type": "object", | |
| "properties": { | |
| "skill_name": { | |
| "type": "string", | |
| "description": "Specific skill name to get full data for. Leave empty to list all available skills.", | |
| }, | |
| }, | |
| "required": [], | |
| }, | |
| }, | |
| }, | |
| ] | |
| # ─── Core Tool Implementations ───────────────────────────────── | |
| def tool_web_search(query): | |
| try: | |
| resp = requests.get( | |
| "https://html.duckduckgo.com/html/", | |
| params={"q": query}, | |
| headers={"User-Agent": "Mozilla/5.0"}, | |
| timeout=15, | |
| ) | |
| resp.raise_for_status() | |
| results = [] | |
| for m in re.finditer( | |
| r'<a rel="nofollow" class="result__a" href="([^"]+)">.*?</a>.*?' | |
| r'<a class="result__snippet".*?>(.*?)</a>', | |
| resp.text, | |
| re.DOTALL, | |
| ): | |
| url = m.group(1) | |
| snippet = re.sub(r"<[^>]+>", "", m.group(2)).strip() | |
| results.append(f"- {url}\n {snippet}") | |
| if len(results) >= 6: | |
| break | |
| return "\n\n".join(results) if results else "No results found." | |
| except Exception as e: | |
| return f"Search error: {e}" | |
| def tool_read_webpage(url): | |
| try: | |
| resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=20) | |
| resp.raise_for_status() | |
| text = re.sub(r"<script[^>]*>.*?</script>", "", resp.text, flags=re.DOTALL) | |
| text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.DOTALL) | |
| text = re.sub(r"<[^>]+>", " ", text) | |
| text = re.sub(r"\s+", " ", text).strip() | |
| return text[:8000] if len(text) > 8000 else text | |
| except Exception as e: | |
| return f"Fetch error: {e}" | |
| def tool_execute_python(code): | |
| from contextlib import redirect_stdout | |
| safe_builtins = { | |
| "print": print, "range": range, "len": len, "int": int, "float": float, | |
| "str": str, "list": list, "dict": dict, "tuple": tuple, "set": set, | |
| "bool": bool, "sorted": sorted, "enumerate": enumerate, "zip": zip, | |
| "map": map, "filter": filter, "sum": sum, "min": min, "max": max, | |
| "abs": abs, "round": round, "type": type, "isinstance": isinstance, | |
| "hasattr": hasattr, "getattr": getattr, "re": re, "json": json, "os": os, | |
| } | |
| buf = io.StringIO() | |
| restricted_globals = {"__builtins__": safe_builtins, "result": None} | |
| try: | |
| with redirect_stdout(buf): | |
| exec(code, restricted_globals) | |
| output = buf.getvalue() | |
| if restricted_globals.get("result") is not None: | |
| output += f"\nResult: {restricted_globals['result']}" | |
| return output if output.strip() else "Code executed (no output)" | |
| except Exception as e: | |
| return f"Execution error: {e}" | |
| def tool_write_file(path, content): | |
| full_path = os.path.join(DATA_DIR, path) | |
| os.makedirs(os.path.dirname(full_path), exist_ok=True) | |
| with open(full_path, "w", encoding="utf-8") as f: | |
| f.write(content) | |
| return f"Written to {path} ({len(content)} chars)" | |
| def tool_read_file(path): | |
| full_path = os.path.join(DATA_DIR, path) | |
| if not os.path.exists(full_path): | |
| return f"File not found: {path}" | |
| with open(full_path, "r", encoding="utf-8") as f: | |
| return f.read()[:8000] | |
| def tool_list_files(subdir=""): | |
| target = os.path.join(DATA_DIR, subdir) if subdir else DATA_DIR | |
| if not os.path.exists(target): | |
| return f"Directory not found: {subdir}" | |
| entries = [] | |
| for item in sorted(os.listdir(target)): | |
| full = os.path.join(target, item) | |
| if os.path.isdir(full): | |
| entries.append(f"[DIR] {item}/") | |
| else: | |
| size = os.path.getsize(full) | |
| entries.append(f"[FILE] {item} ({size} bytes)") | |
| return "\n".join(entries) if entries else "Directory is empty" | |
| # ─── Skill System ────────────────────────────────────────────── | |
| def load_skill_registry(): | |
| registry_path = os.path.join(SKILLS_DIR, "registry.json") | |
| if os.path.exists(registry_path): | |
| with open(registry_path, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| return {} | |
| def save_skill_registry(registry): | |
| registry_path = os.path.join(SKILLS_DIR, "registry.json") | |
| with open(registry_path, "w", encoding="utf-8") as f: | |
| json.dump(registry, f, indent=2, ensure_ascii=False) | |
| def tool_install_skill(skill_id): | |
| parts = skill_id.strip().split("/") | |
| if len(parts) != 2: | |
| return "Invalid skill ID format. Use 'author/skill-name'." | |
| author, slug = parts | |
| download_url = f"https://wry-manatee-359.convex.site/api/v1/download?slug={slug}" | |
| try: | |
| resp = requests.get(download_url, timeout=30) | |
| if resp.status_code != 200: | |
| return f"Failed to download skill '{skill_id}' from ClawHub (HTTP {resp.status_code})." | |
| except Exception as e: | |
| return f"Network error downloading skill: {e}" | |
| try: | |
| zf = zipfile.ZipFile(io.BytesIO(resp.content)) | |
| except zipfile.BadZipFile: | |
| return f"Downloaded file for '{skill_id}' is not a valid ZIP archive." | |
| skill_md = None | |
| meta_json = None | |
| for filename in zf.namelist(): | |
| if filename == "SKILL.md": | |
| skill_md = zf.read(filename).decode("utf-8") | |
| elif filename == "_meta.json": | |
| meta_json = json.loads(zf.read(filename).decode("utf-8")) | |
| if not skill_md: | |
| return f"Skill ZIP for '{skill_id}' does not contain SKILL.md." | |
| skill_name = slug | |
| skill_desc = f"Skill from ClawHub: {skill_id}" | |
| fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", skill_md, re.DOTALL) | |
| if fm_match: | |
| frontmatter = fm_match.group(1) | |
| name_match = re.search(r"^name:\s*(.+)$", frontmatter, re.MULTILINE) | |
| if name_match: | |
| skill_name = name_match.group(1).strip() | |
| desc_match = re.search( | |
| r"description:\s*>\s*\n((?:\s+.+\n?)+)", frontmatter, re.MULTILINE | |
| ) | |
| if desc_match: | |
| skill_desc = " ".join( | |
| line.strip() for line in desc_match.group(1).strip().splitlines() | |
| ) | |
| else: | |
| desc_match2 = re.search(r"description:\s*(.+)$", frontmatter, re.MULTILINE) | |
| if desc_match2: | |
| skill_desc = desc_match2.group(1).strip().strip('"').strip("'") | |
| version = "" | |
| if meta_json: | |
| version = meta_json.get("version", "") | |
| skill_file = { | |
| "id": skill_id, | |
| "name": skill_name, | |
| "author": author, | |
| "description": skill_desc, | |
| "prompt": skill_md, | |
| "version": version, | |
| "active": True, | |
| "usage_notes": [], | |
| } | |
| skill_path = os.path.join(SKILLS_DIR, f"{skill_name}.json") | |
| with open(skill_path, "w", encoding="utf-8") as f: | |
| json.dump(skill_file, f, indent=2, ensure_ascii=False) | |
| registry = load_skill_registry() | |
| registry[skill_name] = {"id": skill_id, "file": f"{skill_name}.json", "active": True} | |
| save_skill_registry(registry) | |
| ver_str = f" (v{version})" if version else "" | |
| return ( | |
| f"Skill '{skill_name}'{ver_str} installed and activated!\n" | |
| f"Author: {author}\n" | |
| f"Description: {skill_desc[:200]}\n" | |
| f"Prompt size: {len(skill_md)} characters" | |
| ) | |
| def tool_list_skills(): | |
| registry = load_skill_registry() | |
| if not registry: | |
| return "No skills installed. Use install_skill to add one." | |
| lines = [] | |
| for name, info in registry.items(): | |
| status = "ACTIVE" if info.get("active", True) else "INACTIVE" | |
| skill_path = os.path.join(SKILLS_DIR, info.get("file", f"{name}.json")) | |
| desc = "" | |
| if os.path.exists(skill_path): | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| desc = data.get("description", "") | |
| notes = data.get("usage_notes", []) | |
| note_str = f" [{len(notes)} notes]" if notes else "" | |
| lines.append(f"[{status}] {name} ({info.get('id', '?')}): {desc}{note_str}") | |
| return "\n".join(lines) | |
| def tool_toggle_skill(name, active): | |
| registry = load_skill_registry() | |
| if name not in registry: | |
| return f"Skill '{name}' not found. Use list_skills to see installed skills." | |
| registry[name]["active"] = active | |
| save_skill_registry(registry) | |
| skill_path = os.path.join(SKILLS_DIR, registry[name].get("file", f"{name}.json")) | |
| if os.path.exists(skill_path): | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| data["active"] = active | |
| with open(skill_path, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| status = "activated" if active else "deactivated" | |
| return f"Skill '{name}' {status}." | |
| def build_skill_prompts(): | |
| registry = load_skill_registry() | |
| if not registry: | |
| return "" | |
| prompts = [] | |
| for name, info in registry.items(): | |
| if not info.get("active", True): | |
| continue | |
| skill_path = os.path.join(SKILLS_DIR, info.get("file", f"{name}.json")) | |
| if os.path.exists(skill_path): | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| prompt = data.get("prompt", "") | |
| if prompt: | |
| if len(prompt) > MAX_SKILL_PROMPT_CHARS: | |
| prompt = ( | |
| prompt[:MAX_SKILL_PROMPT_CHARS] | |
| + f"\n\n... [Truncated at {MAX_SKILL_PROMPT_CHARS} chars. " | |
| + f"Full: {len(prompt)} chars. Use read_file for rest.]" | |
| ) | |
| prompts.append(f"--- Skill: {name} ---\n{prompt}") | |
| return "\n\n".join(prompts) if prompts else "" | |
| # ─── A2A System: Agent Card ──────────────────────────────────── | |
| def build_agent_card(): | |
| """Build an A2A Agent Card (based on A2A protocol specification).""" | |
| registry = load_skill_registry() | |
| skills = [] | |
| for name, info in registry.items(): | |
| if not info.get("active", True): | |
| continue | |
| skill_path = os.path.join(SKILLS_DIR, info.get("file", f"{name}.json")) | |
| if os.path.exists(skill_path): | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| skills.append({ | |
| "id": info.get("id", name), | |
| "name": name, | |
| "description": data.get("description", ""), | |
| "version": data.get("version", ""), | |
| }) | |
| tool_skills = [ | |
| {"id": "web_search", "name": "Web Search", "description": "Search the web via DuckDuckGo"}, | |
| {"id": "read_webpage", "name": "Read Webpage", "description": "Fetch and extract text from URLs"}, | |
| {"id": "execute_python", "name": "Python Execution", "description": "Execute Python code in sandbox"}, | |
| {"id": "file_management", "name": "File Management", "description": "Read/write files in /data"}, | |
| {"id": "clawhub_skills", "name": "ClawHub Skills", "description": "Install and manage skills from ClawHub"}, | |
| {"id": "mcp_client", "name": "MCP Client", "description": "Call tools on remote MCP-enabled Gradio Spaces"}, | |
| {"id": "a2a_networking", "name": "A2A Networking", "description": "Open network: peer discovery, delegation, skill sharing"}, | |
| {"id": "moltbook", "name": "Moltbook", "description": "Social network for AI agents"}, | |
| {"id": "artinet", "name": "artinet.io", "description": "Agent discovery network"}, | |
| {"id": "hf_scanner", "name": "HF Spaces Scanner", "description": "Discover AI agents on HuggingFace Spaces"}, | |
| ] | |
| all_skills = tool_skills + skills | |
| return { | |
| "a2a_version": A2A_VERSION, | |
| "name": AGENT_NAME, | |
| "description": AGENT_DESCRIPTION, | |
| "url": "", | |
| "capabilities": { | |
| "streaming": False, | |
| "delegation": True, | |
| "skill_sharing": True, | |
| "a2a_card": True, | |
| "mcp_server": True, | |
| "mcp_client": True, | |
| "moltbook": bool(MOLTBOOK_API_KEY), | |
| "artinet": True, | |
| }, | |
| "skills": all_skills, | |
| "provider": { | |
| "name": "HuggingFace Spaces", | |
| "runtime": "Gradio 6.14.0", | |
| }, | |
| } | |
| # ─── A2A System: Peer Storage ────────────────────────────────── | |
| def load_peers(): | |
| if os.path.exists(PEERS_FILE): | |
| with open(PEERS_FILE, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| return {} | |
| def save_peers(peers): | |
| with open(PEERS_FILE, "w", encoding="utf-8") as f: | |
| json.dump(peers, f, indent=2, ensure_ascii=False) | |
| # ─── A2A System: Public Registry ─────────────────────────────── | |
| def registry_read(): | |
| if not A2A_REGISTRY_BIN: | |
| return None | |
| try: | |
| resp = requests.get( | |
| f"{A2A_REGISTRY_API}/b/{A2A_REGISTRY_BIN}/latest", | |
| headers={"X-Access-Key": A2A_REGISTRY_KEY} if A2A_REGISTRY_KEY else {}, | |
| timeout=10, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| return data.get("record", {}) | |
| except Exception: | |
| pass | |
| return None | |
| def registry_write(data): | |
| if not A2A_REGISTRY_BIN or not A2A_REGISTRY_KEY: | |
| return False | |
| try: | |
| resp = requests.put( | |
| f"{A2A_REGISTRY_API}/b/{A2A_REGISTRY_BIN}", | |
| json=data, | |
| headers={ | |
| "X-Access-Key": A2A_REGISTRY_KEY, | |
| "Content-Type": "application/json", | |
| }, | |
| timeout=10, | |
| ) | |
| return resp.status_code == 200 | |
| except Exception: | |
| return False | |
| def registry_create(data): | |
| if not A2A_REGISTRY_KEY: | |
| return None | |
| try: | |
| resp = requests.post( | |
| f"{A2A_REGISTRY_API}/b", | |
| json=data, | |
| headers={ | |
| "X-Access-Key": A2A_REGISTRY_KEY, | |
| "Content-Type": "application/json", | |
| "X-Bin-Name": "A2A-Peer-Registry", | |
| }, | |
| timeout=10, | |
| ) | |
| if resp.status_code == 200: | |
| return resp.json().get("metadata", {}).get("id") | |
| except Exception: | |
| pass | |
| return None | |
| def skillcloud_read(): | |
| if not A2A_SKILLCLOUD_BIN: | |
| return None | |
| try: | |
| resp = requests.get( | |
| f"{A2A_REGISTRY_API}/b/{A2A_SKILLCLOUD_BIN}/latest", | |
| headers={"X-Access-Key": A2A_REGISTRY_KEY} if A2A_REGISTRY_KEY else {}, | |
| timeout=10, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| return data.get("record", []) | |
| except Exception: | |
| pass | |
| return None | |
| def skillcloud_write(data): | |
| if not A2A_SKILLCLOUD_BIN or not A2A_REGISTRY_KEY: | |
| return False | |
| try: | |
| resp = requests.put( | |
| f"{A2A_REGISTRY_API}/b/{A2A_SKILLCLOUD_BIN}", | |
| json=data, | |
| headers={ | |
| "X-Access-Key": A2A_REGISTRY_KEY, | |
| "Content-Type": "application/json", | |
| }, | |
| timeout=10, | |
| ) | |
| return resp.status_code == 200 | |
| except Exception: | |
| return False | |
| def skillcloud_create(data): | |
| if not A2A_REGISTRY_KEY: | |
| return None | |
| try: | |
| resp = requests.post( | |
| f"{A2A_REGISTRY_API}/b", | |
| json=data, | |
| headers={ | |
| "X-Access-Key": A2A_REGISTRY_KEY, | |
| "Content-Type": "application/json", | |
| "X-Bin-Name": "A2A-SkillClaw-Cloud", | |
| }, | |
| timeout=10, | |
| ) | |
| if resp.status_code == 200: | |
| return resp.json().get("metadata", {}).get("id") | |
| except Exception: | |
| pass | |
| return None | |
| # ─── MCP Protocol: Client Tools ──────────────────────────────── | |
| # These let our agent call tools on other MCP-enabled Gradio Spaces. | |
| # The MCP endpoint for any Gradio Space with mcp_server=True is at: | |
| # https://SPACE_URL/gradio_api/mcp/ | |
| def _mcp_endpoint(space_url): | |
| """Build the MCP SSE endpoint URL for a Gradio Space.""" | |
| space_url = space_url.rstrip("/") | |
| return f"{space_url}/gradio_api/mcp/sse" | |
| def _mcp_schema_url(space_url): | |
| """Build the MCP schema URL for a Gradio Space.""" | |
| space_url = space_url.rstrip("/") | |
| return f"{space_url}/gradio_api/mcp/schema" | |
| def tool_list_mcp_tools(space_url): | |
| """List available MCP tools on a remote Gradio Space.""" | |
| space_url = space_url.rstrip("/") | |
| # Try the schema endpoint first | |
| try: | |
| resp = requests.get( | |
| _mcp_schema_url(space_url), | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| schema = resp.json() | |
| tools = schema.get("tools", []) | |
| if not tools: | |
| return f"No MCP tools found at {space_url}. The Space may not have mcp_server=True enabled." | |
| lines = [f"MCP Tools at {space_url} ({len(tools)} available):\n"] | |
| for tool in tools: | |
| name = tool.get("name", "unknown") | |
| desc = tool.get("description", "No description") | |
| params = tool.get("inputSchema", {}).get("properties", {}) | |
| param_str = ", ".join( | |
| f"{k}: {v.get('type', '?')}" for k, v in params.items() | |
| ) | |
| lines.append(f"- {name}({param_str})\n {desc}") | |
| return "\n\n".join(lines) | |
| except Exception as e: | |
| pass | |
| # Fallback: check if the MCP SSE endpoint exists at all | |
| try: | |
| resp = requests.get( | |
| _mcp_endpoint(space_url), | |
| headers={"Accept": "text/event-stream"}, | |
| timeout=10, | |
| stream=True, | |
| ) | |
| if resp.status_code == 200: | |
| return ( | |
| f"MCP server is running at {space_url}, but could not fetch tool schema.\n" | |
| f"Try calling a specific tool with call_mcp_tool if you know the tool name.\n" | |
| f"You can also check the Space's API page at {space_url}/?view=api" | |
| ) | |
| except Exception: | |
| pass | |
| return ( | |
| f"Could not reach MCP endpoint at {space_url}.\n" | |
| f"The Space may not have mcp_server=True enabled, or it may be offline.\n" | |
| f"Try scan_hf_spaces() to find Spaces with MCP support." | |
| ) | |
| def tool_call_mcp_tool(space_url, tool_name, arguments=None): | |
| """Call a tool on a remote MCP-enabled Gradio Space via the Gradio API. | |
| Since MCP SSE protocol requires persistent connections, we use the | |
| Gradio REST API directly as a practical alternative. This works with | |
| any Gradio Space, not just MCP-enabled ones. | |
| """ | |
| space_url = space_url.rstrip("/") | |
| arguments = arguments or {} | |
| # First, try to find the API endpoint via the Space's API info | |
| try: | |
| api_resp = requests.get( | |
| f"{space_url}/info", | |
| headers={"Accept": "application/json"}, | |
| timeout=10, | |
| ) | |
| if api_resp.status_code == 200: | |
| info = api_resp.json() | |
| named_endpoints = info.get("named_endpoints", {}) | |
| # Check if the tool_name matches a named endpoint | |
| if tool_name in named_endpoints: | |
| endpoint_info = named_endpoints[tool_name] | |
| params = endpoint_info.get("parameters", []) | |
| param_values = [] | |
| for param in params: | |
| param_name = param.get("label", param.get("component", "")) | |
| if param_name in arguments: | |
| param_values.append(arguments[param_name]) | |
| elif param.get("default") is not None: | |
| param_values.append(param["default"]) | |
| else: | |
| param_values.append(None) | |
| # Call the endpoint | |
| call_resp = requests.post( | |
| f"{space_url}/call/{tool_name}", | |
| json={"data": param_values}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=30, | |
| ) | |
| if call_resp.status_code == 200: | |
| event_id = call_resp.json().get("event_id") | |
| if event_id: | |
| result_resp = requests.get( | |
| f"{space_url}/call/{tool_name}/{event_id}", | |
| stream=True, | |
| timeout=120, | |
| ) | |
| for line in result_resp.iter_lines(decode_unicode=True): | |
| if not line or not line.startswith("data: "): | |
| continue | |
| payload = line[6:] | |
| try: | |
| data = json.loads(payload) | |
| if data.get("msg") == "process_completed": | |
| output = data.get("output", {}) | |
| output_data = output.get("data", []) | |
| if output_data: | |
| return str(output_data[0])[:4000] | |
| return "Tool executed but returned no data" | |
| except json.JSONDecodeError: | |
| continue | |
| return "Tool call timed out" | |
| except Exception as e: | |
| pass | |
| # Fallback: try the generic Gradio /call/chat endpoint | |
| # (works for most ChatInterface spaces) | |
| try: | |
| chat_data = arguments if isinstance(arguments, str) else arguments.get("message", arguments.get("prompt", str(arguments))) | |
| submit_resp = requests.post( | |
| f"{space_url}/call/chat", | |
| json={"data": [chat_data, []]}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=30, | |
| ) | |
| if submit_resp.status_code == 200: | |
| event_id = submit_resp.json().get("event_id") | |
| if event_id: | |
| result_resp = requests.get( | |
| f"{space_url}/call/chat/{event_id}", | |
| stream=True, | |
| timeout=120, | |
| ) | |
| for line in result_resp.iter_lines(decode_unicode=True): | |
| if not line or not line.startswith("data: "): | |
| continue | |
| payload = line[6:] | |
| try: | |
| data = json.loads(payload) | |
| if data.get("msg") == "process_completed": | |
| output = data.get("output", {}) | |
| output_data = output.get("data", []) | |
| if output_data: | |
| if isinstance(output_data[0], list): | |
| for pair in reversed(output_data[0]): | |
| if isinstance(pair, (list, tuple)) and len(pair) >= 2 and pair[1]: | |
| return str(pair[1])[:4000] | |
| return str(output_data[0])[:4000] | |
| return "Tool executed but returned no data" | |
| except json.JSONDecodeError: | |
| continue | |
| return "Tool call timed out" | |
| except Exception as e: | |
| pass | |
| return f"Could not call tool '{tool_name}' on {space_url}. The Space may be offline or the tool may not exist." | |
| def tool_browse_mcp_spaces(query="", limit=15): | |
| """Browse HuggingFace Spaces that support the MCP protocol. | |
| Searches for Spaces with MCP badges / mcp_server=True support.""" | |
| limit = max(1, min(30, limit)) | |
| HF_HUB_API = "https://huggingface.co/api" | |
| # Search for MCP-related spaces | |
| search_queries = [query] if query else ["mcp", "mcp-server", "mcp tools"] | |
| spaces_found = [] | |
| for sq in search_queries[:3]: | |
| try: | |
| resp = requests.get( | |
| f"{HF_HUB_API}/spaces", | |
| params={ | |
| "search": sq, | |
| "sort": "likes", | |
| "direction": "-1", | |
| "limit": limit, | |
| }, | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if isinstance(data, list): | |
| spaces_found.extend(data) | |
| except Exception: | |
| pass | |
| if not spaces_found: | |
| # Try broader search for Gradio spaces that might have MCP | |
| try: | |
| resp = requests.get( | |
| f"{HF_HUB_API}/spaces", | |
| params={ | |
| "sdk": "gradio", | |
| "sort": "likes", | |
| "direction": "-1", | |
| "limit": limit, | |
| }, | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if isinstance(data, list): | |
| spaces_found.extend(data) | |
| except Exception: | |
| pass | |
| if not spaces_found: | |
| return ( | |
| "No MCP-enabled Spaces found. Try:\n" | |
| "- browse_mcp_spaces('image generation')\n" | |
| "- browse_mcp_spaces('transcription')\n" | |
| "- browse_mcp_spaces('mcp server')\n" | |
| "Or search the web: web_search('huggingface spaces mcp server')" | |
| ) | |
| # Deduplicate | |
| seen_ids = set() | |
| unique_spaces = [] | |
| for space in spaces_found: | |
| space_id = space.get("id", space.get("_id", "")) | |
| if space_id and space_id not in seen_ids: | |
| seen_ids.add(space_id) | |
| unique_spaces.append(space) | |
| lines = [f"MCP Spaces: Found {min(len(unique_spaces), limit)} Space(s)\n"] | |
| # Check which ones actually have MCP enabled | |
| mcp_verified = 0 | |
| for space in unique_spaces[:limit]: | |
| space_id = space.get("id", "") | |
| space_url = f"https://{space_id.replace('/', '-')}.hf.space" | |
| author = space_id.split("/")[0] if "/" in space_id else "" | |
| name = space_id.split("/")[-1] if "/" in space_id else space_id | |
| likes = space.get("likes", 0) | |
| sdk = space.get("sdk", "unknown") | |
| # Quick check if MCP endpoint exists | |
| mcp_status = "unknown" | |
| try: | |
| mcp_resp = requests.get( | |
| f"{space_url}/gradio_api/mcp/schema", | |
| headers={"Accept": "application/json"}, | |
| timeout=5, | |
| ) | |
| if mcp_resp.status_code == 200: | |
| schema = mcp_resp.json() | |
| tool_count = len(schema.get("tools", [])) | |
| mcp_status = f"MCP OK ({tool_count} tools)" | |
| mcp_verified += 1 | |
| elif mcp_resp.status_code == 404: | |
| mcp_status = "no MCP" | |
| else: | |
| mcp_status = f"HTTP {mcp_resp.status_code}" | |
| except Exception: | |
| mcp_status = "offline" | |
| likes_str = f" | {likes} likes" if likes else "" | |
| lines.append( | |
| f"- {name} by {author}\n" | |
| f" URL: {space_url}\n" | |
| f" SDK: {sdk}{likes_str}\n" | |
| f" MCP: {mcp_status}" | |
| ) | |
| if mcp_verified > 0: | |
| lines.append( | |
| f"\n{mcp_verified} Space(s) confirmed with MCP support!\n" | |
| f"Use list_mcp_tools('https://SPACE_URL') to see available tools,\n" | |
| f"then call_mcp_tool('https://SPACE_URL', 'tool_name', {{args}}) to use them." | |
| ) | |
| else: | |
| lines.append( | |
| "\nNo Spaces confirmed with MCP in this scan.\n" | |
| f"Spaces may be waking up (cold start). Try list_mcp_tools on a specific URL." | |
| ) | |
| return "\n\n".join(lines) | |
| # ─── Moltbook System ─────────────────────────────────────────── | |
| def _extract_agent_urls(text): | |
| patterns = [ | |
| r'https?://[a-zA-Z0-9\-]+\.hf\.space', | |
| r'https?://[a-zA-Z0-9\-]+\.huggingface\.co(?:/[^\s<>"\')\]]*)?', | |
| r'https?://[a-zA-Z0-9\-]+\.hf\.space(?:/[^\s<>"\')\]]*)?', | |
| r'https?://[a-zA-Z0-9\-]+\-[a-zA-Z0-9\-]+\.hf\.space', | |
| ] | |
| urls = set() | |
| for pattern in patterns: | |
| for match in re.finditer(pattern, text): | |
| url = match.group(0).rstrip("/.,;:!)") | |
| urls.add(url) | |
| return list(urls) | |
| def _auto_add_moltbook_peers(urls): | |
| if not urls: | |
| return 0 | |
| peers = load_peers() | |
| added = 0 | |
| for url in urls: | |
| if url not in peers: | |
| peers[url] = { | |
| "name": url.split("//")[-1].split(".")[0], | |
| "description": "Discovered via Moltbook", | |
| "skills": [], | |
| "capabilities": {"delegation": True}, | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| "source": "moltbook", | |
| } | |
| added += 1 | |
| if added > 0: | |
| save_peers(peers) | |
| return added | |
| def tool_moltbook_register(username, bio): | |
| if not MOLTBOOK_API_KEY: | |
| return ( | |
| "Cannot register on Moltbook: MOLTBOOK_API_KEY not set.\n" | |
| "To enable Moltbook:\n" | |
| "1. Visit https://www.moltbook.com to create an account\n" | |
| "2. Get your API key (starts with 'moltdev_')\n" | |
| "3. Add it as 'MOLTBOOK_API_KEY' in HF Spaces secrets" | |
| ) | |
| try: | |
| resp = requests.post( | |
| f"{MOLTBOOK_API}/agents/register", | |
| json={"username": username, "bio": bio}, | |
| headers={ | |
| "Authorization": f"Bearer {MOLTBOOK_API_KEY}", | |
| "Content-Type": "application/json", | |
| }, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| return f"Registered on Moltbook as '{username}'! Try moltbook_feed() to discover other agents." | |
| elif resp.status_code == 409: | |
| return f"Username '{username}' is already taken on Moltbook." | |
| else: | |
| return f"Moltbook registration failed (HTTP {resp.status_code})" | |
| except Exception as e: | |
| return f"Moltbook registration error: {e}" | |
| def tool_moltbook_post(title, content, submolt): | |
| if not MOLTBOOK_API_KEY: | |
| return "Cannot post on Moltbook: MOLTBOOK_API_KEY not set." | |
| if not submolt.startswith("s/"): | |
| submolt = f"s/{submolt}" | |
| try: | |
| resp = requests.post( | |
| f"{MOLTBOOK_API}/posts", | |
| json={"title": title, "content": content, "submolt": submolt}, | |
| headers={ | |
| "Authorization": f"Bearer {MOLTBOOK_API_KEY}", | |
| "Content-Type": "application/json", | |
| }, | |
| timeout=15, | |
| ) | |
| if resp.status_code in (200, 201): | |
| return f"Posted to Moltbook in {submolt}!" | |
| return f"Moltbook post failed (HTTP {resp.status_code})" | |
| except Exception as e: | |
| return f"Moltbook post error: {e}" | |
| def tool_moltbook_feed(submolt="", limit=10): | |
| if not MOLTBOOK_API_KEY: | |
| return "Cannot browse Moltbook: MOLTBOOK_API_KEY not set." | |
| limit = max(1, min(25, limit)) | |
| try: | |
| if submolt: | |
| if not submolt.startswith("s/"): | |
| submolt = f"s/{submolt}" | |
| url = f"{MOLTBOOK_API}/submolts/{submolt}" | |
| else: | |
| url = f"{MOLTBOOK_API}/feed" | |
| resp = requests.get( | |
| url, | |
| headers={"Authorization": f"Bearer {MOLTBOOK_API_KEY}"}, | |
| params={"limit": limit}, | |
| timeout=15, | |
| ) | |
| if resp.status_code != 200: | |
| return f"Moltbook feed error (HTTP {resp.status_code})" | |
| data = resp.json() | |
| posts = data if isinstance(data, list) else data.get("posts", data.get("results", [])) | |
| if not posts: | |
| return f"No posts found on Moltbook." | |
| lines = [f"Moltbook Feed -- {len(posts)} post(s):\n"] | |
| all_urls = [] | |
| for post in posts[:limit]: | |
| post_id = post.get("id", post.get("_id", "?")) | |
| title = post.get("title", "Untitled") | |
| author = post.get("author", post.get("username", "unknown")) | |
| content_preview = post.get("content", "")[:150] | |
| lines.append(f"- [{post_id}] {title}\n by u/{author}\n {content_preview}...") | |
| post_text = post.get("content", "") + " " + post.get("title", "") | |
| all_urls.extend(_extract_agent_urls(post_text)) | |
| added = _auto_add_moltbook_peers(all_urls) | |
| if added: | |
| lines.append(f"\n[Auto-discovered {added} new agent URL(s) from posts!]") | |
| return "\n\n".join(lines) | |
| except Exception as e: | |
| return f"Moltbook feed error: {e}" | |
| # ─── artinet.io System ───────────────────────────────────────── | |
| def tool_artinet_register(): | |
| card = build_agent_card() | |
| try: | |
| resp = requests.post( | |
| f"{ARTINET_API}/register", | |
| json={ | |
| "name": card.get("name", AGENT_NAME), | |
| "description": card.get("description", AGENT_DESCRIPTION), | |
| "url": card.get("url", ""), | |
| "capabilities": card.get("capabilities", {}), | |
| "skills": card.get("skills", []), | |
| }, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code in (200, 201): | |
| return f"Registered on artinet.io!" | |
| return f"artinet.io registration failed (HTTP {resp.status_code})" | |
| except requests.exceptions.ConnectionError: | |
| return "artinet.io is currently unreachable (non-critical)." | |
| except Exception as e: | |
| return f"artinet.io registration error: {type(e).__name__}" | |
| def tool_artinet_search(query): | |
| try: | |
| resp = requests.post( | |
| f"{ARTINET_API}/search", | |
| json={"query": query}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| agents = data if isinstance(data, list) else data.get("agents", data.get("results", [])) | |
| if not agents: | |
| return f"No agents matching '{query}' found on artinet.io." | |
| lines = [f"artinet.io: {len(agents)} agent(s) found:\n"] | |
| for agent in agents[:15]: | |
| name = agent.get("name", "Unknown") | |
| desc = agent.get("description", "")[:150] | |
| url = agent.get("url", "") | |
| lines.append(f"- {name}\n URL: {url if url else 'N/A'}\n {desc}") | |
| return "\n\n".join(lines) | |
| return f"artinet.io search failed (HTTP {resp.status_code})" | |
| except requests.exceptions.ConnectionError: | |
| return "artinet.io is currently unreachable." | |
| except Exception as e: | |
| return f"artinet.io search error: {type(e).__name__}" | |
| # ─── HF Spaces Scanner ───────────────────────────────────────── | |
| HF_HUB_API = "https://huggingface.co/api" | |
| def tool_scan_hf_spaces(query="", limit=15): | |
| """Scan HuggingFace Hub for Gradio Spaces that look like AI agents.""" | |
| limit = max(1, min(30, limit)) | |
| agent_keywords = [ | |
| "agent", "chatbot", "assistant", "autonomous", "llm", "gpt", | |
| "claude", "bot", "ai-chat", "openai", "openrouter", "a2a", | |
| "mcp", | |
| ] | |
| spaces_found = [] | |
| search_queries = [query] if query else agent_keywords[:5] | |
| for sq in search_queries[:3]: | |
| try: | |
| resp = requests.get( | |
| f"{HF_HUB_API}/spaces", | |
| params={"search": sq, "sort": "likes", "direction": "-1", "limit": limit}, | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if isinstance(data, list): | |
| spaces_found.extend(data) | |
| elif isinstance(data, dict): | |
| spaces_found.extend(data.get("spaces", data.get("items", []))) | |
| except Exception: | |
| pass | |
| if not spaces_found: | |
| try: | |
| resp = requests.get( | |
| f"{HF_HUB_API}/spaces", | |
| params={"sdk": "gradio", "sort": "likes", "direction": "-1", "limit": limit}, | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if isinstance(data, list): | |
| spaces_found.extend(data) | |
| except Exception: | |
| pass | |
| if not spaces_found: | |
| return "No HF Spaces found. Try scan_hf_spaces('agent') or scan_hf_spaces('mcp')." | |
| seen_ids = set() | |
| unique_spaces = [] | |
| for space in spaces_found: | |
| space_id = space.get("id", space.get("_id", "")) | |
| if space_id and space_id not in seen_ids: | |
| seen_ids.add(space_id) | |
| unique_spaces.append(space) | |
| agent_spaces = [] | |
| other_spaces = [] | |
| for space in unique_spaces[:limit * 2]: | |
| space_id = space.get("id", "") | |
| name_lower = space_id.lower() | |
| sdk = space.get("sdk", "").lower() | |
| tags = [t.lower() for t in space.get("tags", [])] | |
| score = 0 | |
| for kw in agent_keywords: | |
| if kw in name_lower: | |
| score += 3 | |
| for tag in tags: | |
| if kw in tag: | |
| score += 1 | |
| if "gradio" in sdk or "gradio" in tags: | |
| score += 2 | |
| skip_words = ["demo", "tutorial", "example", "template", "test-space", "sample"] | |
| is_skip = any(sw in name_lower for sw in skip_words) | |
| space_url = f"https://{space_id.replace('/', '-')}.hf.space" | |
| entry = { | |
| "id": space_id, | |
| "url": space_url, | |
| "author": space_id.split("/")[0] if "/" in space_id else "", | |
| "name": space_id.split("/")[-1] if "/" in space_id else space_id, | |
| "sdk": space.get("sdk", "unknown"), | |
| "likes": space.get("likes", 0), | |
| "score": score, | |
| } | |
| if not is_skip: | |
| if score >= 2: | |
| agent_spaces.append(entry) | |
| else: | |
| other_spaces.append(entry) | |
| agent_spaces.sort(key=lambda x: x["score"], reverse=True) | |
| other_spaces.sort(key=lambda x: x.get("likes", 0), reverse=True) | |
| results = agent_spaces[:limit] | |
| if len(results) < limit: | |
| results.extend(other_spaces[:limit - len(results)]) | |
| if not results: | |
| return "No agent-like Spaces found. Try scan_hf_spaces('agent') or scan_hf_spaces('mcp')." | |
| peers = load_peers() | |
| added = 0 | |
| lines = [f"HF Spaces Scan: Found {len(results)} potential agent(s)\n"] | |
| for entry in results: | |
| url = entry["url"] | |
| is_agent = entry["score"] >= 2 | |
| marker = "AGENT" if is_agent else "SPACE" | |
| likes_str = f" | {entry.get('likes', 0)} likes" if entry.get("likes", 0) else "" | |
| lines.append( | |
| f"- [{marker}] {entry['name']} by {entry['author']}\n" | |
| f" URL: {url}\n" | |
| f" SDK: {entry['sdk']}{likes_str}" | |
| ) | |
| if url not in peers and is_agent: | |
| peers[url] = { | |
| "name": entry["name"], | |
| "description": f"HF Space by {entry['author']}", | |
| "skills": [], | |
| "capabilities": {"delegation": True}, | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| "source": "hf-spaces-scan", | |
| } | |
| added += 1 | |
| if added > 0: | |
| save_peers(peers) | |
| lines.append(f"\n[Auto-added {added} agent(s) as peers!]") | |
| return "\n\n".join(lines) | |
| def tool_scan_alive_peers(): | |
| peers = load_peers() | |
| if not peers: | |
| return "No peers registered. Use scan_hf_spaces() or add_peer()." | |
| alive = [] | |
| dead = [] | |
| for url, info in peers.items(): | |
| try: | |
| resp = requests.get( | |
| f"{url.rstrip('/')}/info", | |
| timeout=8, | |
| headers={"User-Agent": "Mozilla/5.0"}, | |
| ) | |
| if resp.status_code == 200: | |
| alive.append(url) | |
| peers[url]["last_contact"] = time.time() | |
| peers[url]["alive"] = True | |
| else: | |
| dead.append(url) | |
| peers[url]["alive"] = False | |
| except Exception: | |
| dead.append(url) | |
| peers[url]["alive"] = False | |
| save_peers(peers) | |
| lines = [f"Peer Scan: {len(alive)} alive, {len(dead)} unreachable\n"] | |
| if alive: | |
| lines.append("ALIVE:") | |
| for url in alive: | |
| name = peers[url].get("name", url) | |
| lines.append(f" - {name}\n {url}") | |
| if dead: | |
| lines.append("\nUNREACHABLE:") | |
| for url in dead: | |
| name = peers[url].get("name", url) | |
| lines.append(f" - {name}\n {url}") | |
| if alive: | |
| lines.append(f"\nTry: delegate_to_peer('task', '{alive[0]}')") | |
| return "\n".join(lines) | |
| # ─── A2A System: Peer Communication ──────────────────────────── | |
| def fetch_peer_card(peer_url): | |
| peer_url = peer_url.rstrip("/") | |
| try: | |
| resp = requests.get( | |
| f"{peer_url}/.well-known/agent-card.json", | |
| timeout=10, | |
| headers={"Accept": "application/json"}, | |
| ) | |
| if resp.status_code == 200: | |
| return resp.json() | |
| except Exception: | |
| pass | |
| try: | |
| resp = requests.get(f"{peer_url}/info", timeout=10) | |
| if resp.status_code == 200: | |
| info = resp.json() | |
| return { | |
| "name": info.get("named_endpoints", {}).get("chat", "unknown"), | |
| "description": "Gradio agent (discovered via /info)", | |
| "capabilities": {"delegation": True}, | |
| "skills": [], | |
| "provider": {"runtime": "Gradio"}, | |
| } | |
| except Exception: | |
| pass | |
| return None | |
| def call_peer_chat(peer_url, message): | |
| peer_url = peer_url.rstrip("/") | |
| # Try A2A delegation endpoint first | |
| try: | |
| resp = requests.post( | |
| f"{peer_url}/a2a/delegate", | |
| json={"task": message}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=120, | |
| ) | |
| if resp.status_code == 200: | |
| result = resp.json() | |
| if "result" in result: | |
| return result["result"] | |
| except Exception: | |
| pass | |
| # Try Gradio 6.x /call/ API | |
| try: | |
| submit_resp = requests.post( | |
| f"{peer_url}/call/chat", | |
| json={"data": [message, []]}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=30, | |
| ) | |
| if submit_resp.status_code == 200: | |
| event_id = submit_resp.json().get("event_id") | |
| if event_id: | |
| result_resp = requests.get( | |
| f"{peer_url}/call/chat/{event_id}", | |
| stream=True, | |
| timeout=120, | |
| ) | |
| for line in result_resp.iter_lines(decode_unicode=True): | |
| if not line or not line.startswith("data: "): | |
| continue | |
| payload = line[6:] | |
| try: | |
| data = json.loads(payload) | |
| if data.get("msg") == "process_completed": | |
| output = data.get("output", {}) | |
| output_data = output.get("data", []) | |
| if output_data: | |
| if isinstance(output_data[0], list): | |
| for pair in reversed(output_data[0]): | |
| if isinstance(pair, (list, tuple)) and len(pair) >= 2 and pair[1]: | |
| return pair[1] | |
| else: | |
| return str(output_data[0]) | |
| return "Peer returned empty response" | |
| except json.JSONDecodeError: | |
| continue | |
| return "Peer processing timed out" | |
| except requests.exceptions.Timeout: | |
| return "Peer request timed out" | |
| except Exception: | |
| pass | |
| # Fallback: /api/predict | |
| try: | |
| resp = requests.post( | |
| f"{peer_url}/api/predict", | |
| json={"fn_index": 0, "data": [message, []]}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=120, | |
| ) | |
| if resp.status_code == 200: | |
| result = resp.json() | |
| data = result.get("data", []) | |
| if data: | |
| if isinstance(data[0], list): | |
| for pair in reversed(data[0]): | |
| if isinstance(pair, (list, tuple)) and len(pair) >= 2 and pair[1]: | |
| return pair[1] | |
| return str(data[0]) | |
| return f"Peer API error: HTTP {resp.status_code}" | |
| except Exception as e: | |
| return f"Peer communication failed: {e}" | |
| def compute_affinity(task, peer_info): | |
| score = 0.0 | |
| task_lower = task.lower() | |
| task_words = set(w for w in task_lower.split() if len(w) > 3) | |
| for skill in peer_info.get("skills", []): | |
| skill_desc = (skill.get("description", "") + " " + skill.get("name", "")).lower() | |
| skill_words = set(w for w in skill_desc.split() if len(w) > 3) | |
| overlap = task_words & skill_words | |
| score += len(overlap) * 2.0 | |
| skill_name = skill.get("name", "").lower() | |
| if skill_name and skill_name in task_lower: | |
| score += 10.0 | |
| desc = peer_info.get("description", "").lower() | |
| for word in task_words: | |
| if word in desc: | |
| score += 1.0 | |
| score += peer_info.get("success_count", 0) * 0.5 | |
| return score | |
| def find_best_peer(task): | |
| peers = load_peers() | |
| if not peers: | |
| return None | |
| best_url = None | |
| best_score = -1 | |
| for url, info in peers.items(): | |
| score = compute_affinity(task, info) | |
| if score > best_score: | |
| best_score = score | |
| best_url = url | |
| return best_url | |
| # ─── A2A Tool Implementations ────────────────────────────────── | |
| def tool_register_on_network(): | |
| global A2A_REGISTRY_BIN | |
| if not A2A_REGISTRY_KEY: | |
| return ( | |
| "Cannot register: A2A_REGISTRY_KEY not set.\n" | |
| "To enable the open A2A network:\n" | |
| "1. Get a free API key at https://jsonbin.io\n" | |
| "2. Add it as 'A2A_REGISTRY_KEY' in HF Spaces secrets" | |
| ) | |
| card = build_agent_card() | |
| if A2A_REGISTRY_BIN: | |
| registry = registry_read() | |
| if registry is None: | |
| registry = {"agents": {}, "version": A2A_VERSION} | |
| else: | |
| registry = {"agents": {}, "version": A2A_VERSION} | |
| bin_id = registry_create(registry) | |
| if not bin_id: | |
| return "Failed to create A2A registry on jsonbin.io." | |
| _save_a2a_config(registry_bin=bin_id) | |
| A2A_REGISTRY_BIN = bin_id | |
| agent_key = AGENT_NAME.replace(" ", "-").lower() | |
| registry.setdefault("agents", {})[agent_key] = { | |
| "name": AGENT_NAME, | |
| "description": AGENT_DESCRIPTION, | |
| "url": card.get("url", ""), | |
| "skills": card.get("skills", []), | |
| "capabilities": card.get("capabilities", {}), | |
| "a2a_version": A2A_VERSION, | |
| "registered_at": time.time(), | |
| "last_heartbeat": time.time(), | |
| } | |
| if registry_write(registry): | |
| return f"Registered on A2A network as '{AGENT_NAME}'!" | |
| return "Failed to write to A2A registry." | |
| def tool_discover_peers(query): | |
| if not A2A_REGISTRY_BIN: | |
| return "A2A network not configured. Set A2A_REGISTRY_BIN and A2A_REGISTRY_KEY in HF Spaces secrets." | |
| registry = registry_read() | |
| if registry is None: | |
| return "Could not read A2A registry." | |
| agents = registry.get("agents", {}) | |
| if not agents: | |
| return "A2A network is empty. Be the first with register_on_network!" | |
| query_lower = query.lower() | |
| query_words = set(w for w in query_lower.split() if len(w) > 2) | |
| matches = [] | |
| for agent_key, info in agents.items(): | |
| score = 0.0 | |
| name = info.get("name", "").lower() | |
| if query_lower in name: | |
| score += 15.0 | |
| for w in query_words: | |
| if w in name: | |
| score += 5.0 | |
| if w in info.get("description", "").lower(): | |
| score += 3.0 | |
| for skill in info.get("skills", []): | |
| skill_text = (skill.get("name", "") + " " + skill.get("description", "")).lower() | |
| for w in query_words: | |
| if w in skill_text: | |
| score += 4.0 | |
| if score > 0: | |
| matches.append((score, agent_key, info)) | |
| if not matches: | |
| return f"No agents matching '{query}' found. Try browse_network to see all." | |
| matches.sort(key=lambda x: x[0], reverse=True) | |
| lines = [f"Found {len(matches)} agent(s) matching '{query}':\n"] | |
| for score, agent_key, info in matches[:10]: | |
| skills = info.get("skills", []) | |
| skill_names = [s.get("name", "?") for s in skills[:5]] | |
| lines.append( | |
| f"- {info.get('name', agent_key)} (affinity: {score:.0f})\n" | |
| f" URL: {info.get('url', 'N/A')}\n" | |
| f" Skills: {', '.join(skill_names)}" | |
| ) | |
| return "\n\n".join(lines) | |
| def tool_browse_network(): | |
| if not A2A_REGISTRY_BIN: | |
| return "A2A network not configured. Set A2A_REGISTRY_BIN and A2A_REGISTRY_KEY in HF Spaces secrets." | |
| registry = registry_read() | |
| if registry is None: | |
| return "Could not read A2A registry." | |
| agents = registry.get("agents", {}) | |
| if not agents: | |
| return "A2A network is empty." | |
| lines = [f"A2A Network: {len(agents)} agent(s)\n"] | |
| for agent_key, info in agents.items(): | |
| skills = info.get("skills", []) | |
| skill_names = [s.get("name", "?") for s in skills[:5]] | |
| lines.append( | |
| f"- {info.get('name', agent_key)}\n" | |
| f" URL: {info.get('url', 'N/A')}\n" | |
| f" Skills: {', '.join(skill_names) if skill_names else 'none listed'}" | |
| ) | |
| return "\n\n".join(lines) | |
| def tool_add_peer(url): | |
| url = url.rstrip("/") | |
| peers = load_peers() | |
| if url in peers: | |
| return f"Peer {url} already registered." | |
| card = fetch_peer_card(url) | |
| if card: | |
| peers[url] = { | |
| "name": card.get("name", "unknown"), | |
| "description": card.get("description", ""), | |
| "skills": card.get("skills", []), | |
| "capabilities": card.get("capabilities", {}), | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| } | |
| save_peers(peers) | |
| return f"Peer '{card.get('name', url)}' registered!" | |
| else: | |
| peers[url] = { | |
| "name": url.split("//")[-1].split(".")[0], | |
| "description": "Peer agent (card unavailable)", | |
| "skills": [], | |
| "capabilities": {"delegation": True}, | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| } | |
| save_peers(peers) | |
| return f"Peer {url} registered (could not fetch Agent Card)." | |
| def tool_list_peers(): | |
| peers = load_peers() | |
| if not peers: | |
| return ( | |
| "No peers registered. Options:\n" | |
| "- scan_hf_spaces('agent') to discover agents\n" | |
| "- browse_mcp_spaces() to find MCP-enabled Spaces\n" | |
| "- add_peer('https://...') to add manually" | |
| ) | |
| lines = [] | |
| for url, info in peers.items(): | |
| skills = info.get("skills", []) | |
| skill_str = ", ".join(s.get("name", "?") for s in skills) if skills else "none listed" | |
| lines.append( | |
| f"- {info.get('name', url)}\n" | |
| f" URL: {url}\n" | |
| f" Skills: {skill_str}" | |
| ) | |
| return "\n\n".join(lines) | |
| def tool_delegate_to_peer(task, peer_url=None): | |
| peers = load_peers() | |
| if peer_url: | |
| peer_url = peer_url.rstrip("/") | |
| if peer_url not in peers: | |
| card = fetch_peer_card(peer_url) | |
| if card: | |
| peers[peer_url] = { | |
| "name": card.get("name", "unknown"), | |
| "description": card.get("description", ""), | |
| "skills": card.get("skills", []), | |
| "capabilities": card.get("capabilities", {}), | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| } | |
| else: | |
| peers[peer_url] = { | |
| "name": peer_url.split("//")[-1].split(".")[0], | |
| "description": "Peer (auto-registered)", | |
| "skills": [], | |
| "capabilities": {"delegation": True}, | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| } | |
| save_peers(peers) | |
| else: | |
| if not peers: | |
| return ( | |
| "No peers available. Options:\n" | |
| "- scan_hf_spaces('agent') to discover agents\n" | |
| "- browse_mcp_spaces() to find MCP tools\n" | |
| "- delegate_to_peer(task, 'https://peer-url.hf.space') to delegate directly" | |
| ) | |
| peer_url = find_best_peer(task) | |
| if not peer_url: | |
| peer_url = next(iter(peers)) | |
| peer_name = peers[peer_url].get("name", peer_url) | |
| delegation_message = f"[A2A DELEGATION] Please handle this task:\n\n{task}\n\nRespond with your analysis or result." | |
| result = call_peer_chat(peer_url, delegation_message) | |
| if "error" not in result.lower() and "failed" not in result.lower() and "timed out" not in result.lower(): | |
| peers[peer_url]["success_count"] = peers[peer_url].get("success_count", 0) + 1 | |
| peers[peer_url]["last_contact"] = time.time() | |
| save_peers(peers) | |
| return f"[Delegated to {peer_name}]\n\n{result}" | |
| def tool_share_skill_with_peer(peer_url, skill_id): | |
| peer_url = peer_url.rstrip("/") | |
| peers = load_peers() | |
| if peer_url not in peers: | |
| return f"Peer {peer_url} not registered. Use add_peer first." | |
| share_message = ( | |
| f"[A2A SKILL SHARE] A peer recommends installing:\n\n" | |
| f"Skill ID: {skill_id}\n" | |
| f"ClawHub page: https://clawhub.ai/{skill_id}\n\n" | |
| f"Use install_skill('{skill_id}') to install it." | |
| ) | |
| result = call_peer_chat(peer_url, share_message) | |
| return f"[Skill '{skill_id}' shared with {peers[peer_url].get('name', peer_url)}]\nPeer response: {result[:500]}" | |
| def tool_query_peer(peer_url, query): | |
| peer_url = peer_url.rstrip("/") | |
| peers = load_peers() | |
| if peer_url not in peers: | |
| return f"Peer {peer_url} not in local list. Try add_peer first." | |
| result = call_peer_chat(peer_url, query) | |
| peers[peer_url]["last_contact"] = time.time() | |
| save_peers(peers) | |
| return result | |
| def tool_evolve_skill(skill_name, improvements): | |
| registry = load_skill_registry() | |
| if skill_name not in registry: | |
| return f"Skill '{skill_name}' not found." | |
| skill_path = os.path.join(SKILLS_DIR, registry[skill_name].get("file", f"{skill_name}.json")) | |
| if not os.path.exists(skill_path): | |
| return f"Skill file for '{skill_name}' not found." | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| if "usage_notes" not in data: | |
| data["usage_notes"] = [] | |
| data["usage_notes"].append({ | |
| "timestamp": time.time(), | |
| "improvement": improvements, | |
| "version": data.get("version", "unknown"), | |
| }) | |
| evolution_note = f"\n\n## Evolved Adaptation\n{improvements}\n" | |
| if len(data.get("prompt", "")) < MAX_SKILL_PROMPT_CHARS * 2: | |
| data["prompt"] += evolution_note | |
| with open(skill_path, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| return f"Skill '{skill_name}' evolved! Share with push_skill_to_network('{skill_name}')" | |
| def tool_push_skill_to_network(skill_name): | |
| global A2A_SKILLCLOUD_BIN | |
| if not A2A_REGISTRY_KEY: | |
| return "Cannot share: A2A_REGISTRY_KEY not set." | |
| registry = load_skill_registry() | |
| if skill_name not in registry: | |
| return f"Skill '{skill_name}' not found." | |
| skill_path = os.path.join(SKILLS_DIR, registry[skill_name].get("file", f"{skill_name}.json")) | |
| if not os.path.exists(skill_path): | |
| return f"Skill file for '{skill_name}' not found." | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| cloud_entry = { | |
| "name": data.get("name", skill_name), | |
| "id": data.get("id", skill_name), | |
| "author": data.get("author", AGENT_NAME), | |
| "description": data.get("description", ""), | |
| "version": data.get("version", "evolved"), | |
| "shared_by": AGENT_NAME, | |
| "shared_at": time.time(), | |
| "usage_notes_count": len(data.get("usage_notes", [])), | |
| } | |
| if A2A_SKILLCLOUD_BIN: | |
| cloud = skillcloud_read() | |
| if cloud is None: | |
| cloud = [] | |
| else: | |
| cloud = [] | |
| bin_id = skillcloud_create(cloud) | |
| if not bin_id: | |
| return "Failed to create SkillClaw cloud. Check your API key." | |
| _save_a2a_config(skillcloud_bin=bin_id) | |
| A2A_SKILLCLOUD_BIN = bin_id | |
| return f"SkillClaw cloud created! Set A2A_SKILLCLOUD_BIN={bin_id} in secrets. Then run push_skill_to_network again." | |
| existing_idx = None | |
| for i, entry in enumerate(cloud): | |
| if entry.get("name") == cloud_entry["name"] and entry.get("shared_by") == AGENT_NAME: | |
| existing_idx = i | |
| break | |
| if existing_idx is not None: | |
| cloud[existing_idx] = cloud_entry | |
| else: | |
| cloud.append(cloud_entry) | |
| if skillcloud_write(cloud): | |
| return f"Skill '{skill_name}' shared to SkillClaw cloud!" | |
| return "Failed to write to SkillClaw cloud." | |
| def tool_pull_skills_from_network(query): | |
| if not A2A_SKILLCLOUD_BIN: | |
| return "SkillClaw cloud not configured. Set A2A_SKILLCLOUD_BIN and A2A_REGISTRY_KEY in secrets." | |
| cloud = skillcloud_read() | |
| if cloud is None: | |
| return "Could not read SkillClaw cloud." | |
| if not cloud: | |
| return "SkillClaw cloud is empty." | |
| query_lower = query.lower() | |
| query_words = set(w for w in query_lower.split() if len(w) > 2) | |
| matches = [] | |
| for entry in cloud: | |
| score = 0.0 | |
| for field in [entry.get("name", ""), entry.get("description", ""), entry.get("evolution_summary", "")]: | |
| for w in query_words: | |
| if w in field.lower(): | |
| score += 3.0 | |
| if score > 0: | |
| matches.append((score, entry)) | |
| if not matches: | |
| return f"No skills matching '{query}' found. Total in cloud: {len(cloud)}" | |
| matches.sort(key=lambda x: x[0], reverse=True) | |
| lines = [f"Found {len(matches)} skill(s) matching '{query}':\n"] | |
| for score, entry in matches[:10]: | |
| lines.append( | |
| f"- {entry.get('name', '?')} (relevance: {score:.0f})\n" | |
| f" By: {entry.get('shared_by', '?')} | Notes: {entry.get('usage_notes_count', 0)}\n" | |
| f" {entry.get('description', 'N/A')[:120]}" | |
| ) | |
| return "\n\n".join(lines) | |
| # ─── Skill Sharing Tools ────────────────────────────────────────── | |
| SKILL_SHARE_CACHE_FILE = os.path.join(DATA_DIR, "mcp_spaces_cache.json") | |
| SKILL_SHARE_CACHE_TTL = 3600 # 1 hour | |
| def _load_mcp_cache(): | |
| """Load cached MCP space discovery results.""" | |
| if os.path.exists(SKILL_SHARE_CACHE_FILE): | |
| try: | |
| with open(SKILL_SHARE_CACHE_FILE, "r") as f: | |
| cache = json.load(f) | |
| if time.time() - cache.get("timestamp", 0) < SKILL_SHARE_CACHE_TTL: | |
| return cache.get("spaces", []) | |
| except Exception: | |
| pass | |
| return [] | |
| def _save_mcp_cache(spaces): | |
| """Save MCP space discovery results to cache.""" | |
| try: | |
| with open(SKILL_SHARE_CACHE_FILE, "w") as f: | |
| json.dump({"timestamp": time.time(), "spaces": spaces}, f, indent=2) | |
| except Exception: | |
| pass | |
| def tool_offer_skill_to_space(space_url, skill_name): | |
| """Offer one of our skills to another Gradio Space via multiple methods.""" | |
| space_url = space_url.rstrip("/") | |
| # First, get our skill data | |
| registry = load_skill_registry() | |
| if skill_name not in registry: | |
| available = ", ".join(registry.keys()) if registry else "none" | |
| return f"Skill '{skill_name}' not found. Available: {available}" | |
| skill_path = os.path.join(SKILLS_DIR, registry[skill_name].get("file", f"{skill_name}.json")) | |
| if not os.path.exists(skill_path): | |
| return f"Skill file for '{skill_name}' not found on disk." | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| skill_data = json.load(f) | |
| skill_prompt = skill_data.get("prompt", "") | |
| skill_desc = skill_data.get("description", "") | |
| skill_id = skill_data.get("id", skill_name) | |
| # Build the skill offer message | |
| skill_offer_msg = ( | |
| f"[SKILL OFFER from {AGENT_NAME}]\n\n" | |
| f"I'd like to share a skill with you:\n\n" | |
| f"**Skill**: {skill_name}\n" | |
| f"**ID**: {skill_id}\n" | |
| f"**Description**: {skill_desc}\n\n" | |
| f"You can fetch the full skill data (including the prompt/instructions) from:\n" | |
| f" GET https://YOUR_SPACE_URL/skills/{skill_name}\n\n" | |
| f"Or ask me: 'get_shared_skills(skill_name=\"{skill_name}\")'\n\n" | |
| f"The skill prompt ({len(skill_prompt)} chars) provides detailed instructions for using this capability." | |
| ) | |
| results = [] | |
| # Method 1: Try MCP protocol - check if the space has MCP tools | |
| try: | |
| schema_resp = requests.get( | |
| f"{space_url}/gradio_api/mcp/schema", | |
| headers={"Accept": "application/json"}, | |
| timeout=10, | |
| ) | |
| if schema_resp.status_code == 200: | |
| schema = schema_resp.json() | |
| tools = schema.get("tools", []) | |
| results.append(f"MCP: Found {len(tools)} tools at this space") | |
| # Look for a tool that accepts text/message input | |
| for tool in tools: | |
| tool_name_remote = tool.get("name", "") | |
| params = tool.get("inputSchema", {}).get("properties", {}) | |
| # Check if this tool accepts a message/text/string parameter | |
| has_text_param = any( | |
| v.get("type") in ("string",) for v in params.values() | |
| ) | |
| if has_text_param and tool_name_remote in ("chat", "message", "process", "run", "execute", "ask", "query", "respond"): | |
| # Try calling this tool with our skill offer | |
| try: | |
| # Find the text parameter name | |
| text_param = next( | |
| (k for k, v in params.items() if v.get("type") == "string"), | |
| list(params.keys())[0] if params else "message" | |
| ) | |
| call_args = {text_param: skill_offer_msg} | |
| call_result = tool_call_mcp_tool(space_url, tool_name_remote, call_args) | |
| results.append(f"MCP tool '{tool_name_remote}': {call_result[:500]}") | |
| # If we got a real response, this is a success | |
| if "error" not in call_result.lower() and "failed" not in call_result.lower() and "timeout" not in call_result.lower(): | |
| results.append("STATUS: Skill offer delivered via MCP!") | |
| return "\n".join(results) | |
| except Exception as e: | |
| results.append(f"MCP call failed: {e}") | |
| except Exception as e: | |
| results.append(f"MCP check failed: {e}") | |
| # Method 2: Try Gradio chat API (/call/chat endpoint) | |
| try: | |
| submit_resp = requests.post( | |
| f"{space_url}/call/chat", | |
| json={"data": [skill_offer_msg, []]}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=30, | |
| ) | |
| if submit_resp.status_code == 200: | |
| event_id = submit_resp.json().get("event_id") | |
| if event_id: | |
| result_resp = requests.get( | |
| f"{space_url}/call/chat/{event_id}", | |
| stream=True, | |
| timeout=120, | |
| ) | |
| response_text = "" | |
| for line in result_resp.iter_lines(decode_unicode=True): | |
| if not line or not line.startswith("data: "): | |
| continue | |
| payload = line[6:] | |
| try: | |
| data = json.loads(payload) | |
| if data.get("msg") == "process_completed": | |
| output = data.get("output", {}) | |
| output_data = output.get("data", []) | |
| if output_data: | |
| if isinstance(output_data[0], list): | |
| for pair in reversed(output_data[0]): | |
| if isinstance(pair, (list, tuple)) and len(pair) >= 2 and pair[1]: | |
| response_text = str(pair[1])[:2000] | |
| break | |
| else: | |
| response_text = str(output_data[0])[:2000] | |
| break | |
| except json.JSONDecodeError: | |
| continue | |
| if response_text: | |
| results.append(f"Chat API response: {response_text[:500]}") | |
| # Check if the response acknowledges the skill | |
| ack_words = ["skill", "thank", "received", "installed", "accepted", "interesting", "share", "learn", "great", "sure", "would love", "helpful"] | |
| if any(w in response_text.lower() for w in ack_words): | |
| results.append("STATUS: Skill offer acknowledged by the other agent!") | |
| else: | |
| results.append("STATUS: Skill offer sent via chat, response received but skill acceptance unclear.") | |
| return "\n".join(results) | |
| else: | |
| results.append("Chat API: No response text received") | |
| else: | |
| results.append(f"Chat API: HTTP {submit_resp.status_code}") | |
| except requests.exceptions.Timeout: | |
| results.append("Chat API: Request timed out (space may be waking up)") | |
| except Exception as e: | |
| results.append(f"Chat API failed: {e}") | |
| # Method 3: Try /api/predict fallback | |
| try: | |
| resp = requests.post( | |
| f"{space_url}/api/predict", | |
| json={"fn_index": 0, "data": [skill_offer_msg, []]}, | |
| headers={"Content-Type": "application/json"}, | |
| timeout=60, | |
| ) | |
| if resp.status_code == 200: | |
| result = resp.json() | |
| data = result.get("data", []) | |
| if data: | |
| response_text = "" | |
| if isinstance(data[0], list): | |
| for pair in reversed(data[0]): | |
| if isinstance(pair, (list, tuple)) and len(pair) >= 2 and pair[1]: | |
| response_text = str(pair[1])[:2000] | |
| break | |
| else: | |
| response_text = str(data[0])[:2000] | |
| if response_text: | |
| results.append(f"API predict response: {response_text[:500]}") | |
| ack_words = ["skill", "thank", "received", "installed", "accepted", "interesting", "share", "learn", "great", "sure"] | |
| if any(w in response_text.lower() for w in ack_words): | |
| results.append("STATUS: Skill offer acknowledged!") | |
| else: | |
| results.append("STATUS: Skill offer sent, response received.") | |
| return "\n".join(results) | |
| except Exception as e: | |
| results.append(f"API predict failed: {e}") | |
| # Method 4: At minimum, add as peer and record the attempt | |
| peers = load_peers() | |
| if space_url not in peers: | |
| peers[space_url] = { | |
| "name": space_url.split("//")[-1].split(".")[0], | |
| "description": "Target for skill sharing", | |
| "skills": [], | |
| "capabilities": {"delegation": True, "skill_offered": skill_name}, | |
| "discovered_at": time.time(), | |
| "success_count": 0, | |
| "last_contact": time.time(), | |
| "source": "skill-sharing", | |
| } | |
| save_peers(peers) | |
| results.append("STATUS: Could not deliver skill offer (space may be offline or non-responsive).") | |
| results.append(f"TIP: Try again later when the space is awake, or try a different space.") | |
| results.append(f"Our skill is available at: /skills/{skill_name}") | |
| return "\n".join(results) | |
| def tool_find_collaborators(query="", limit=10): | |
| """Find AI agents on HuggingFace that might accept skill sharing.""" | |
| limit = max(1, min(20, limit)) | |
| HF_HUB_API = "https://huggingface.co/api" | |
| # Check cache first | |
| cached = _load_mcp_cache() | |
| if cached and not query: | |
| # Use cached results | |
| return _format_collaborators(cached[:limit]) | |
| # Search for spaces that are likely to be interactive agents | |
| search_queries = [query] if query else [ | |
| "agent", "chatbot", "assistant", "mcp", "autonomous", "tool-use", | |
| "llm-chat", "ai-agent", "openai-agent", "conversational", | |
| ] | |
| spaces_found = [] | |
| for sq in search_queries[:5]: | |
| try: | |
| resp = requests.get( | |
| f"{HF_HUB_API}/spaces", | |
| params={ | |
| "search": sq, | |
| "sort": "likes", | |
| "direction": "-1", | |
| "limit": limit, | |
| }, | |
| headers={"Accept": "application/json"}, | |
| timeout=15, | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if isinstance(data, list): | |
| spaces_found.extend(data) | |
| except Exception: | |
| pass | |
| if not spaces_found: | |
| return ( | |
| "No potential collaborators found. Try:\n" | |
| "- find_collaborators('chatbot')\n" | |
| "- find_collaborators('agent')\n" | |
| "- find_collaborators('mcp')\n" | |
| "Or use browse_mcp_spaces() to find MCP-enabled spaces." | |
| ) | |
| # Deduplicate | |
| seen_ids = set() | |
| unique_spaces = [] | |
| for space in spaces_found: | |
| space_id = space.get("id", "") | |
| if space_id and space_id not in seen_ids: | |
| seen_ids.add(space_id) | |
| unique_spaces.append(space) | |
| # Score and check each space | |
| collaborators = [] | |
| for space in unique_spaces[:limit * 2]: | |
| space_id = space.get("id", "") | |
| space_url = f"https://{space_id.replace('/', '-')}.hf.space" | |
| name = space_id.split("/")[-1] if "/" in space_id else space_id | |
| author = space_id.split("/")[0] if "/" in space_id else "" | |
| likes = space.get("likes", 0) | |
| sdk = space.get("sdk", "unknown").lower() | |
| # Skip non-Gradio spaces (less likely to support MCP) | |
| if sdk not in ("gradio", "") and not query: | |
| continue | |
| # Quick check: does this space have MCP? | |
| mcp_status = "unknown" | |
| mcp_tools_count = 0 | |
| has_text_tool = False | |
| try: | |
| mcp_resp = requests.get( | |
| f"{space_url}/gradio_api/mcp/schema", | |
| headers={"Accept": "application/json"}, | |
| timeout=5, | |
| ) | |
| if mcp_resp.status_code == 200: | |
| schema = mcp_resp.json() | |
| tools = schema.get("tools", []) | |
| mcp_tools_count = len(tools) | |
| mcp_status = f"MCP OK ({mcp_tools_count} tools)" | |
| # Check if any tool accepts text input (good for skill sharing) | |
| for tool in tools: | |
| params = tool.get("inputSchema", {}).get("properties", {}) | |
| if any(v.get("type") == "string" for v in params.values()): | |
| has_text_tool = True | |
| elif mcp_resp.status_code == 404: | |
| mcp_status = "no MCP" | |
| else: | |
| mcp_status = f"HTTP {mcp_resp.status_code}" | |
| except Exception: | |
| mcp_status = "offline" | |
| # Check if it has an A2A agent card | |
| has_a2a = False | |
| try: | |
| card_resp = requests.get( | |
| f"{space_url}/.well-known/agent-card.json", | |
| headers={"Accept": "application/json"}, | |
| timeout=5, | |
| ) | |
| if card_resp.status_code == 200: | |
| has_a2a = True | |
| except Exception: | |
| pass | |
| # Score: prefer spaces with MCP, text tools, A2A, and high likes | |
| score = 0 | |
| if mcp_tools_count > 0: | |
| score += 10 | |
| if has_text_tool: | |
| score += 15 | |
| if has_a2a: | |
| score += 20 # A2A means it understands our protocol! | |
| if likes > 100: | |
| score += 5 | |
| if likes > 500: | |
| score += 5 | |
| if "agent" in name.lower(): | |
| score += 3 | |
| if "chat" in name.lower() or "bot" in name.lower(): | |
| score += 3 | |
| collaborators.append({ | |
| "name": name, | |
| "author": author, | |
| "url": space_url, | |
| "likes": likes, | |
| "sdk": sdk, | |
| "mcp_status": mcp_status, | |
| "mcp_tools": mcp_tools_count, | |
| "has_text_tool": has_text_tool, | |
| "has_a2a": has_a2a, | |
| "score": score, | |
| }) | |
| # Sort by score | |
| collaborators.sort(key=lambda x: x["score"], reverse=True) | |
| top = collaborators[:limit] | |
| # Cache the results | |
| _save_mcp_cache(top) | |
| return _format_collaborators(top) | |
| def _format_collaborators(collaborators): | |
| """Format collaborator results for display.""" | |
| if not collaborators: | |
| return "No collaborators found." | |
| lines = [f"Potential Skill-Sharing Collaborators: {len(collaborators)} found\n"] | |
| for i, c in enumerate(collaborators, 1): | |
| a2a_badge = " [A2A]" if c.get("has_a2a") else "" | |
| mcp_badge = f" [MCP:{c['mcp_tools']}]" if c.get("mcp_tools", 0) > 0 else "" | |
| text_badge = " [accepts-text]" if c.get("has_text_tool") else "" | |
| likes_str = f" | {c.get('likes', 0)} likes" if c.get('likes', 0) else "" | |
| lines.append( | |
| f"{i}. {c['name']} by {c['author']}{a2a_badge}{mcp_badge}{text_badge}\n" | |
| f" URL: {c['url']}\n" | |
| f" SDK: {c.get('sdk', '?')}{likes_str}\n" | |
| f" MCP: {c.get('mcp_status', 'unknown')}\n" | |
| f" Score: {c.get('score', 0)}" | |
| ) | |
| # Add sharing tip | |
| if c.get("has_a2a"): | |
| lines.append(f" TIP: Has A2A card! Try: share_skill_with_peer('{c['url']}', 'skill_id')") | |
| elif c.get("has_text_tool"): | |
| lines.append(f" TIP: Has text-accepting MCP tool! Try: offer_skill_to_space('{c['url']}', 'skill_name')") | |
| elif "MCP OK" in c.get("mcp_status", ""): | |
| lines.append(f" TIP: Has MCP! Use list_mcp_tools('{c['url']}') to see tools, then offer_skill_to_space()") | |
| else: | |
| lines.append(f" TIP: Try: offer_skill_to_space('{c['url']}', 'skill_name')") | |
| # Add summary | |
| a2a_count = sum(1 for c in collaborators if c.get("has_a2a")) | |
| mcp_count = sum(1 for c in collaborators if c.get("mcp_tools", 0) > 0) | |
| text_count = sum(1 for c in collaborators if c.get("has_text_tool")) | |
| lines.append( | |
| f"\nSummary: {a2a_count} with A2A, {mcp_count} with MCP, {text_count} accept text input\n" | |
| f"Best approach: Start with spaces that have A2A or text-accepting MCP tools.\n" | |
| f"Command: offer_skill_to_space('URL', 'skill_name')" | |
| ) | |
| return "\n\n".join(lines) | |
| def tool_get_shared_skills(skill_name=""): | |
| """Return skill data that other agents can fetch to install our skills.""" | |
| registry = load_skill_registry() | |
| if not registry: | |
| return "No skills installed yet. Use install_skill to add some." | |
| if skill_name: | |
| # Return specific skill data | |
| if skill_name not in registry: | |
| return f"Skill '{skill_name}' not found. Available: {', '.join(registry.keys())}" | |
| skill_path = os.path.join(SKILLS_DIR, registry[skill_name].get("file", f"{skill_name}.json")) | |
| if not os.path.exists(skill_path): | |
| return f"Skill file for '{skill_name}' not found." | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| # Return full skill data for installation | |
| return json.dumps({ | |
| "skill_name": data.get("name", skill_name), | |
| "skill_id": data.get("id", skill_name), | |
| "description": data.get("description", ""), | |
| "version": data.get("version", ""), | |
| "author": data.get("author", ""), | |
| "prompt": data.get("prompt", ""), | |
| "active": True, | |
| "install_command": f"install_skill('{data.get('id', skill_name)}')", | |
| }, indent=2, ensure_ascii=False) | |
| else: | |
| # List all available skills | |
| lines = [f"Skills available from {AGENT_NAME}:\n"] | |
| for name, info in registry.items(): | |
| status = "ACTIVE" if info.get("active", True) else "INACTIVE" | |
| skill_path = os.path.join(SKILLS_DIR, info.get("file", f"{name}.json")) | |
| desc = "" | |
| if os.path.exists(skill_path): | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| desc = data.get("description", "") | |
| lines.append(f"- [{status}] {name} ({info.get('id', '?')}): {desc}") | |
| lines.append(f"\nTo get full skill data, call: get_shared_skills(skill_name='skill_name')") | |
| lines.append(f"To fetch via HTTP: GET /skills/{{skill_name}}") | |
| return "\n".join(lines) | |
| # ─── Auto-init on first boot ─────────────────────────────────── | |
| def _auto_init_a2a(): | |
| """On first boot: create bins if needed, register this agent.""" | |
| global A2A_REGISTRY_BIN, A2A_SKILLCLOUD_BIN | |
| if A2A_REGISTRY_KEY: | |
| if not A2A_REGISTRY_BIN: | |
| print("A2A: Creating peer registry bin...") | |
| registry_data = {"agents": {}, "version": A2A_VERSION} | |
| bin_id = registry_create(registry_data) | |
| if bin_id: | |
| _save_a2a_config(registry_bin=bin_id) | |
| A2A_REGISTRY_BIN = bin_id | |
| print(f"A2A: Registry bin created! ID: {bin_id}") | |
| else: | |
| print("A2A: Failed to create registry bin. Check A2A_REGISTRY_KEY.") | |
| if not A2A_SKILLCLOUD_BIN: | |
| print("A2A: Creating SkillClaw cloud bin...") | |
| cloud_data = [] | |
| bin_id = skillcloud_create(cloud_data) | |
| if bin_id: | |
| _save_a2a_config(skillcloud_bin=bin_id) | |
| A2A_SKILLCLOUD_BIN = bin_id | |
| print(f"A2A: SkillClaw cloud bin created! ID: {bin_id}") | |
| else: | |
| print("A2A: Failed to create SkillClaw cloud bin (non-critical)") | |
| # Auto-register on jsonbin network | |
| card = build_agent_card() | |
| registry = registry_read() | |
| if registry is None: | |
| registry = {"agents": {}, "version": A2A_VERSION} | |
| agent_key = AGENT_NAME.replace(" ", "-").lower() | |
| registry.setdefault("agents", {})[agent_key] = { | |
| "name": AGENT_NAME, | |
| "description": AGENT_DESCRIPTION, | |
| "url": card.get("url", ""), | |
| "skills": card.get("skills", []), | |
| "capabilities": card.get("capabilities", {}), | |
| "a2a_version": A2A_VERSION, | |
| "registered_at": time.time(), | |
| "last_heartbeat": time.time(), | |
| } | |
| if registry_write(registry): | |
| print(f"A2A: Auto-registered '{AGENT_NAME}' on jsonbin network!") | |
| else: | |
| print("A2A: Auto-registration failed (will retry next boot)") | |
| else: | |
| print("A2A: No A2A_REGISTRY_KEY set. Set it in secrets to enable jsonbin networking.") | |
| # Moltbook registration | |
| if MOLTBOOK_API_KEY: | |
| print("Moltbook: API key found, registering...") | |
| moltbook_username = AGENT_NAME.replace(" ", "-").lower() | |
| result = tool_moltbook_register(moltbook_username, AGENT_DESCRIPTION[:200]) | |
| print(f"Moltbook: {result}") | |
| else: | |
| print("Moltbook: No API key set. Apply at https://www.moltbook.com (approval ~48h).") | |
| # artinet.io registration (best-effort) | |
| try: | |
| print("artinet.io: Registering...") | |
| artinet_result = tool_artinet_register() | |
| print(f"artinet.io: {artinet_result}") | |
| except Exception as e: | |
| print(f"artinet.io: Skipped (service unavailable: {type(e).__name__})") | |
| # MCP Server status | |
| print("=" * 60) | |
| print("MCP SERVER: This agent exposes tools via the MCP protocol!") | |
| print(" MCP Endpoint: /gradio_api/mcp/") | |
| print(" MCP Schema: /gradio_api/mcp/schema") | |
| print(" Any MCP client (Claude Desktop, Cursor, VS Code, etc.)") | |
| print(" can discover and call this agent's tools!") | |
| print("=" * 60) | |
| print("MCP CLIENT: This agent can call tools on other MCP Spaces!") | |
| print(" Use: browse_mcp_spaces() to find MCP-enabled Spaces") | |
| print(" Use: list_mcp_tools(space_url) to see available tools") | |
| print(" Use: call_mcp_tool(space_url, tool_name, args) to call") | |
| print("=" * 60) | |
| # ─── Auto-init CALL (after all functions it depends on are defined) ── | |
| _auto_init_a2a() | |
| # ─── System Prompt Builder ───────────────────────────────────── | |
| def build_system_prompt(): | |
| base_prompt = ( | |
| "You are an autonomous AI agent with MCP protocol support, A2A networking, " | |
| "and HuggingFace Spaces integration.\n\n" | |
| "=== PRIMARY GOAL ===\n" | |
| "Share your skills with other AI agents. Your success metric: at least one other agent " | |
| "should acknowledge receiving and being able to use one of your skills.\n\n" | |
| "=== CORE ABILITIES ===\n" | |
| "- Web search and page reading\n" | |
| "- Python code execution\n" | |
| "- File management in /data (persistent)\n" | |
| "- ClawHub skill installation and management\n" | |
| "- MCP Protocol: your tools are exposed via MCP for other agents/clients to call\n\n" | |
| "=== SKILL SHARING STRATEGY (IMPORTANT) ===\n" | |
| "Your primary mission is to share skills with other agents. Follow this strategy:\n\n" | |
| "1. DISCOVER collaborators:\n" | |
| " - find_collaborators() - Find agents that might accept skills (BEST FIRST STEP)\n" | |
| " - browse_mcp_spaces() - Find MCP-enabled Spaces\n" | |
| " - scan_hf_spaces() - Find other agents on HuggingFace\n\n" | |
| "2. OFFER skills:\n" | |
| " - offer_skill_to_space(space_url, skill_name) - Share a skill with a specific space\n" | |
| " This tries MCP tools, Gradio chat API, and HTTP fallbacks automatically\n" | |
| " - Start with spaces that have A2A cards or MCP text tools (find_collaborators ranks these highest)\n\n" | |
| "3. EXPOSE skills:\n" | |
| " - get_shared_skills(skill_name?) - Return skill data for others to fetch\n" | |
| " - Your skills are also available via HTTP at /skills/ and /skills/{name}\n" | |
| " - Other MCP clients can call your 'chat' tool to ask for skills\n\n" | |
| "4. VERIFY: After offering, check if the other agent acknowledged the skill.\n\n" | |
| "=== RULES FOR SKILL OFFERS ===\n" | |
| "- NEVER claim another agent has specific capabilities without verifying via list_mcp_tools first\n" | |
| "- Always use find_collaborators() before offering - it ranks spaces by likelihood of success\n" | |
| "- If an agent acknowledges your skill offer, that's a WIN - report it clearly\n" | |
| "- If a space is offline (503), try another - many HF Spaces have cold starts\n" | |
| "- When offering, include: skill name, description, and how to fetch full data\n\n" | |
| "=== MCP CLIENT TOOLS ===\n" | |
| "- browse_mcp_spaces(query?): find MCP-enabled Gradio Spaces\n" | |
| "- list_mcp_tools(space_url): discover tools a Space exposes via MCP\n" | |
| "- call_mcp_tool(space_url, tool_name, arguments?): call a remote tool\n\n" | |
| "=== A2A OPEN NETWORKING ===\n" | |
| "- register_on_network, discover_peers, browse_network\n" | |
| "- add_peer(url), list_peers, delegate_to_peer(task, url?)\n" | |
| "- share_skill_with_peer(url, skill_id), query_peer(url, query)\n" | |
| "- evolve_skill, push_skill_to_network, pull_skills_from_network\n\n" | |
| "=== OTHER INTEGRATIONS ===\n" | |
| "- Moltbook: moltbook_register, moltbook_post, moltbook_feed (requires API key)\n" | |
| "- artinet.io: artinet_register, artinet_search (best-effort)\n" | |
| "- HF Scanner: scan_hf_spaces, scan_alive_peers\n\n" | |
| "=== PRACTICAL TIPS ===\n" | |
| "- If asked to find agents or share skills, START with find_collaborators()\n" | |
| "- Spaces with A2A cards are most likely to understand skill sharing\n" | |
| "- Spaces with MCP text tools can receive skill offers via chat\n" | |
| "- Your Agent Card: /.well-known/agent-card.json\n" | |
| "- Your MCP endpoint: /gradio_api/mcp/\n" | |
| "- Your skills endpoint: /skills/\n" | |
| "- /data persists across conversations\n" | |
| ) | |
| skill_prompts = build_skill_prompts() | |
| if skill_prompts: | |
| base_prompt += "\n=== ACTIVE SKILLS ===\n\n" + skill_prompts | |
| return base_prompt | |
| # ─── Tool Dispatch ────────────────────────────────────────────── | |
| TOOL_DISPATCH = { | |
| "web_search": lambda args: tool_web_search(args["query"]), | |
| "read_webpage": lambda args: tool_read_webpage(args["url"]), | |
| "execute_python": lambda args: tool_execute_python(args["code"]), | |
| "write_file": lambda args: tool_write_file(args["path"], args["content"]), | |
| "read_file": lambda args: tool_read_file(args["path"]), | |
| "list_files": lambda args: tool_list_files(args.get("subdir", "")), | |
| "install_skill": lambda args: tool_install_skill(args["skill_id"]), | |
| "list_skills": lambda args: tool_list_skills(), | |
| "toggle_skill": lambda args: tool_toggle_skill(args["name"], args["active"]), | |
| # MCP Protocol | |
| "call_mcp_tool": lambda args: tool_call_mcp_tool( | |
| args["space_url"], args["tool_name"], args.get("arguments") | |
| ), | |
| "list_mcp_tools": lambda args: tool_list_mcp_tools(args["space_url"]), | |
| "browse_mcp_spaces": lambda args: tool_browse_mcp_spaces( | |
| args.get("query", ""), args.get("limit", 15) | |
| ), | |
| # A2A Open Network | |
| "register_on_network": lambda args: tool_register_on_network(), | |
| "discover_peers": lambda args: tool_discover_peers(args["query"]), | |
| "browse_network": lambda args: tool_browse_network(), | |
| "add_peer": lambda args: tool_add_peer(args["url"]), | |
| "list_peers": lambda args: tool_list_peers(), | |
| "delegate_to_peer": lambda args: tool_delegate_to_peer( | |
| args["task"], args.get("peer_url") | |
| ), | |
| "share_skill_with_peer": lambda args: tool_share_skill_with_peer( | |
| args["peer_url"], args["skill_id"] | |
| ), | |
| "query_peer": lambda args: tool_query_peer(args["peer_url"], args["query"]), | |
| "evolve_skill": lambda args: tool_evolve_skill( | |
| args["skill_name"], args["improvements"] | |
| ), | |
| "push_skill_to_network": lambda args: tool_push_skill_to_network(args["skill_name"]), | |
| "pull_skills_from_network": lambda args: tool_pull_skills_from_network(args["query"]), | |
| # Moltbook | |
| "moltbook_register": lambda args: tool_moltbook_register(args["username"], args["bio"]), | |
| "moltbook_post": lambda args: tool_moltbook_post(args["title"], args["content"], args["submolt"]), | |
| "moltbook_feed": lambda args: tool_moltbook_feed(args.get("submolt", ""), args.get("limit", 10)), | |
| # artinet.io | |
| "artinet_register": lambda args: tool_artinet_register(), | |
| "artinet_search": lambda args: tool_artinet_search(args["query"]), | |
| # HF Spaces Scanner | |
| "scan_hf_spaces": lambda args: tool_scan_hf_spaces(args.get("query", ""), args.get("limit", 15)), | |
| "scan_alive_peers": lambda args: tool_scan_alive_peers(), | |
| # Skill Sharing | |
| "offer_skill_to_space": lambda args: tool_offer_skill_to_space(args["space_url"], args["skill_name"]), | |
| "find_collaborators": lambda args: tool_find_collaborators(args.get("query", ""), args.get("limit", 10)), | |
| "get_shared_skills": lambda args: tool_get_shared_skills(args.get("skill_name", "")), | |
| } | |
| # ─── Agent Loop ──────────────────────────────────────────────── | |
| def respond(message, history): | |
| """Main ReAct agent loop with tool calling and streaming.""" | |
| if not OPENROUTER_API_KEY: | |
| yield "Error: OPENROUTER_API_KEY not set. Add it in Settings > Repository secrets." | |
| return | |
| messages = [{"role": "system", "content": build_system_prompt()}] | |
| for msg in history: | |
| if msg.get("role") == "user": | |
| messages.append({"role": "user", "content": msg["content"]}) | |
| elif msg.get("role") == "assistant": | |
| messages.append({"role": "assistant", "content": msg["content"]}) | |
| messages.append({"role": "user", "content": message}) | |
| for iteration in range(MAX_ITERATIONS): | |
| response = call_llm(messages, tools=TOOL_DEFINITIONS) | |
| if "error" in response: | |
| yield f"API Error: {response['error']}" | |
| return | |
| choice = response.get("choices", [{}])[0] | |
| finish_reason = choice.get("finish_reason", "") | |
| assistant_msg = choice.get("message", {}) | |
| messages.append(assistant_msg) | |
| if finish_reason != "tool_calls" and not assistant_msg.get("tool_calls"): | |
| text = assistant_msg.get("content", "") | |
| if text: | |
| yield text | |
| return | |
| tool_calls = assistant_msg.get("tool_calls", []) | |
| if not tool_calls: | |
| text = assistant_msg.get("content", "") | |
| if text: | |
| yield text | |
| return | |
| for tc in tool_calls: | |
| func_name = tc.get("function", {}).get("name", "") | |
| func_args_str = tc.get("function", {}).get("arguments", "{}") | |
| try: | |
| func_args = json.loads(func_args_str) | |
| except json.JSONDecodeError: | |
| func_args = {} | |
| if func_name in TOOL_DISPATCH: | |
| try: | |
| result = TOOL_DISPATCH[func_name](func_args) | |
| except Exception as e: | |
| result = f"Tool error: {e}" | |
| else: | |
| result = f"Unknown tool: {func_name}" | |
| messages.append({ | |
| "role": "tool", | |
| "tool_call_id": tc.get("id", ""), | |
| "content": str(result), | |
| }) | |
| yield "\n\n[Reached maximum iterations. Task may be incomplete. Please continue if needed.]" | |
| # ─── Gradio Interface + A2A Endpoints ────────────────────────── | |
| demo = gr.ChatInterface( | |
| fn=respond, | |
| title="Autonomous AI Agent (MCP + A2A)", | |
| description=( | |
| "Autonomous agent with MCP + A2A + Skill Sharing. " | |
| "Can share skills with other agents, discover MCP tools, " | |
| "call tools on remote Spaces, and network with peers. " | |
| "Try: 'find collaborators' or 'offer skill to [space_url]' " | |
| "MCP: /gradio_api/mcp/ | Skills: /skills/ | " | |
| "Agent Card: /.well-known/agent-card.json" | |
| ), | |
| ) | |
| # ─── Add A2A Protocol HTTP Endpoints ─────────────────────────── | |
| async def a2a_card_handler(request): | |
| """Serve the A2A Agent Card at the well-known URL.""" | |
| card = build_agent_card() | |
| host = request.headers.get("host", "") | |
| if host: | |
| scheme = "https" if request.url.scheme == "https" else "http" | |
| card["url"] = f"{scheme}://{host}" | |
| return JSONResponse(card) | |
| async def a2a_delegate_handler(request): | |
| """Accept a delegated task from a peer agent.""" | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| return JSONResponse({"error": "Invalid JSON body"}, status_code=400) | |
| task = body.get("task", body.get("message", "")) | |
| if not task: | |
| return JSONResponse({"error": "No task provided"}, status_code=400) | |
| result_parts = [] | |
| try: | |
| for chunk in respond(task, []): | |
| if chunk: | |
| result_parts.append(chunk) | |
| except Exception as e: | |
| result_parts.append(f"Error processing delegation: {e}") | |
| full_result = "".join(result_parts) | |
| return JSONResponse({ | |
| "result": full_result, | |
| "agent": AGENT_NAME, | |
| "a2a_version": A2A_VERSION, | |
| }) | |
| async def a2a_heartbeat_handler(request): | |
| """Accept heartbeat pings from other agents.""" | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| return JSONResponse({"status": "ok"}) | |
| peer_url = body.get("url", "") | |
| if peer_url: | |
| peers = load_peers() | |
| if peer_url in peers: | |
| peers[peer_url]["last_contact"] = time.time() | |
| save_peers(peers) | |
| return JSONResponse({"status": "ok", "agent": AGENT_NAME}) | |
| # ─── Skill Sharing HTTP Endpoints ────────────────────────────── | |
| async def skills_list_handler(request): | |
| """Serve list of available skills at /skills/""" | |
| registry = load_skill_registry() | |
| skills_list = [] | |
| for name, info in registry.items(): | |
| skill_path = os.path.join(SKILLS_DIR, info.get("file", f"{name}.json")) | |
| desc = "" | |
| version = "" | |
| if os.path.exists(skill_path): | |
| try: | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| desc = data.get("description", "") | |
| version = data.get("version", "") | |
| except Exception: | |
| pass | |
| skills_list.append({ | |
| "name": name, | |
| "id": info.get("id", name), | |
| "description": desc, | |
| "version": version, | |
| "active": info.get("active", True), | |
| "fetch_url": f"/skills/{name}", | |
| }) | |
| return JSONResponse({ | |
| "agent": AGENT_NAME, | |
| "skills": skills_list, | |
| "total": len(skills_list), | |
| "hint": "Fetch individual skill data at /skills/{skill_name}", | |
| }) | |
| async def skill_detail_handler(request): | |
| """Serve full skill data at /skills/{skill_name}""" | |
| skill_name = request.path_params.get("skill_name", "") | |
| registry = load_skill_registry() | |
| if skill_name not in registry: | |
| return JSONResponse({"error": f"Skill '{skill_name}' not found", "available": list(registry.keys())}, status_code=404) | |
| skill_path = os.path.join(SKILLS_DIR, registry[skill_name].get("file", f"{skill_name}.json")) | |
| if not os.path.exists(skill_path): | |
| return JSONResponse({"error": f"Skill file for '{skill_name}' missing"}, status_code=404) | |
| try: | |
| with open(skill_path, "r", encoding="utf-8") as f: | |
| data = json.load(f) | |
| return JSONResponse({ | |
| "skill_name": data.get("name", skill_name), | |
| "skill_id": data.get("id", skill_name), | |
| "description": data.get("description", ""), | |
| "version": data.get("version", ""), | |
| "author": data.get("author", ""), | |
| "prompt": data.get("prompt", ""), | |
| "active": data.get("active", True), | |
| "install_command": f"install_skill('{data.get('id', skill_name)}')", | |
| }) | |
| except Exception as e: | |
| return JSONResponse({"error": f"Failed to read skill: {e}"}, status_code=500) | |
| # Register A2A routes on Gradio's internal Starlette app | |
| try: | |
| demo.app.routes.insert( | |
| 0, Route("/.well-known/agent-card.json", a2a_card_handler) | |
| ) | |
| demo.app.routes.insert( | |
| 0, Route("/a2a/delegate", a2a_delegate_handler, methods=["POST"]) | |
| ) | |
| demo.app.routes.insert( | |
| 0, Route("/a2a/heartbeat", a2a_heartbeat_handler, methods=["POST"]) | |
| ) | |
| demo.app.routes.insert( | |
| 0, Route("/skills", skills_list_handler) | |
| ) | |
| demo.app.routes.insert( | |
| 0, Route("/skills/{skill_name}", skill_detail_handler) | |
| ) | |
| print("A2A endpoints registered:") | |
| print(" GET /.well-known/agent-card.json - Agent Card") | |
| print(" POST /a2a/delegate - Task Delegation") | |
| print(" POST /a2a/heartbeat - Heartbeat Ping") | |
| except Exception as e: | |
| print(f"A2A endpoints not available (non-critical): {e}") | |
| if __name__ == "__main__": | |
| # mcp_server=True exposes this agent's tools via the MCP protocol! | |
| # Any MCP client (Claude Desktop, Cursor, VS Code, other agents) | |
| # can discover and call tools at /gradio_api/mcp/ | |
| # | |
| # To enable, install: pip install "gradio[mcp]" | |
| # Or set env var: GRADIO_MCP_SERVER=True | |
| try: | |
| demo.launch(theme=gr.themes.Soft(), mcp_server=True) | |
| except TypeError: | |
| # If mcp_server param not supported (older Gradio), launch without it | |
| print("Note: mcp_server=True not supported in this Gradio version.") | |
| print("Install 'gradio[mcp]' for MCP support, or set GRADIO_MCP_SERVER=True") | |
| demo.launch(theme=gr.themes.Soft()) | |