elfatwitterintelligenceagent_search_mentions
Monitor influential Twitter discussions about specific cryptocurrency tokens or blockchain topics. Search using up to 5 keywords to retrieve relevant mentions from smart accounts within a defined time frame.
Instructions
Search for mentions of specific tokens or topics on Twitter. This tool finds discussions about cryptocurrencies, blockchain projects, or other topics of interest. It provides the tweets and mentions of smart accounts (only influential ones) and does not contain all tweets. Use this when you want to understand what influential people are saying about a particular token or topic on Twitter. Each of the search keywords should be one word or phrase. A maximum of 5 keywords are allowed. One key word should be one concept. Never use long sentences or phrases as keywords.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| days_ago | No | Number of days to look back | |
| keywords | Yes | List of keywords to search for | |
| limit | No | Maximum number of results (minimum: 20, maximum: 30) |
Implementation Reference
- mesh_mcp_server/server.py:39-54 (registration)Configuration includes 'ElfaTwitterIntelligenceAgent' in DEFAULT_AGENTS, enabling its tools including 'search_mentions' to be registered as 'elfatwitterintelligenceagent_search_mentions'.# Default supported agents DEFAULT_AGENTS = [ "CoinGeckoTokenInfoAgent", "DexScreenerTokenInfoAgent", "ElfaTwitterIntelligenceAgent", "ExaSearchAgent", "TwitterInfoAgent", "AIXBTProjectInfoAgent", "EtherscanAgent", "EvmTokenInfoAgent", "FundingRateAgent", "UnifaiTokenAnalysisAgent", "YahooFinanceAgent", "ZerionWalletAnalysisAgent" ]
- mesh_mcp_server/server.py:163-223 (registration)Dynamically registers tools from agent metadata using tool_id = f'{agent_id.lower()}_{tool_name}', creating 'elfatwitterintelligenceagent_search_mentions' from ElfaTwitterIntelligenceAgent's search_mentions tool.async def process_tool_metadata(self) -> Dict[str, Dict[str, Any]]: """Process agent metadata and extract tool information. Returns: Dictionary mapping tool IDs to tool information """ agents_metadata = await self.fetch_agent_metadata() tool_registry = {} # Log filtering status if self.supported_agents is not None: logger.info( f"Filtering tools using supported agent list ({len(self.supported_agents)} agents specified)" ) else: logger.info("Loading tools from all available agents (no filter applied)") for agent_id, agent_data in agents_metadata.items(): # Skip agents not in our supported list (if a list is specified) if ( self.supported_agents is not None and agent_id not in self.supported_agents ): continue # Process tools for this agent for tool in agent_data.get("tools", []): if tool.get("type") == "function": function_data = tool.get("function", {}) tool_name = function_data.get("name") if not tool_name: continue # Create a unique tool ID tool_id = f"{agent_id.lower()}_{tool_name}" # Get parameters or create default schema parameters = function_data.get("parameters", {}) if not parameters: parameters = { "type": "object", "properties": {}, "required": [], } # Store tool info tool_registry[tool_id] = { "agent_id": agent_id, "tool_name": tool_name, "description": function_data.get("description", ""), "parameters": parameters, } # Log which agents contributed tools agents_with_tools = set(info["agent_id"] for info in tool_registry.values()) logger.info(f"Loaded tools from agents: {', '.join(sorted(agents_with_tools))}") logger.info(f"Successfully loaded {len(tool_registry)} tools") return tool_registry
- mesh_mcp_server/server.py:224-262 (handler)Handler function that proxies tool execution to the remote Mesh API for the specific agent and tool.async def execute_tool( self, agent_id: str, tool_name: str, tool_arguments: Dict[str, Any] ) -> Dict[str, Any]: """Execute a tool on a mesh agent. Args: agent_id: ID of the agent to execute the tool on tool_name: Name of the tool to execute tool_arguments: Arguments to pass to the tool Returns: Tool execution result Raises: ToolExecutionError: If there's an error executing the tool """ request_data = { "agent_id": agent_id, "input": {"tool": tool_name, "tool_arguments": tool_arguments}, } # Add API key if available if Config.HEURIST_API_KEY: request_data["api_key"] = Config.HEURIST_API_KEY try: result = await call_mesh_api( "mesh_request", method="POST", json=request_data ) return result.get("data", result) # Prefer the 'data' field if it exists except MeshApiError as e: # Re-raise API errors with clearer context raise ToolExecutionError(str(e)) from e except Exception as e: logger.error(f"Error calling {agent_id} tool {tool_name}: {e}") raise ToolExecutionError( f"Failed to call {agent_id} tool {tool_name}: {str(e)}" ) from e
- mesh_mcp_server/server.py:300-319 (handler)MCP call_tool handler that dispatches to execute_tool based on the tool name like 'elfatwitterintelligenceagent_search_mentions', extracting agent and original tool name.@app.call_tool() async def call_tool(name: str, arguments: dict) -> List[types.TextContent]: """Call the specified tool with the given arguments.""" try: if name not in self.tool_registry: raise ValueError(f"Unknown tool: {name}") tool_info = self.tool_registry[name] result = await self.execute_tool( agent_id=tool_info["agent_id"], tool_name=tool_info["tool_name"], tool_arguments=arguments, ) # Convert result to TextContent return [types.TextContent(type="text", text=str(result))] except Exception as e: logger.error(f"Error calling tool {name}: {e}") raise ValueError(f"Failed to call tool {name}: {str(e)}") from e