elfatwitterintelligenceagent_search_account
Analyze Twitter accounts by retrieving engagement metrics, follower growth, and influential mentions. Identify topics and cryptocurrencies frequently discussed with data from ELFA API. Input a username and specify historical activity duration for actionable insights.
Instructions
Search for a Twitter account with both mention search and account statistics. This tool provides engagement metrics, follower growth, and mentions by smart users. It does not contain all tweets, but only those of influential users. It also identifies the topics and cryptocurrencies they frequently discuss. Data comes from ELFA API and can analyze several weeks of historical activity.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| days_ago | No | Number of days to look back for mentions | |
| limit | No | Maximum number of mention results | |
| username | Yes | Twitter username to analyze (without @) |
Implementation Reference
- mesh_mcp_server/server.py:301-319 (handler)MCP handler that executes the tool 'elfatwitterintelligenceagent_search_account' by extracting agent_id='ElfaTwitterIntelligenceAgent' and tool_name='search_account' from the name and proxying to the Mesh API endpoint.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
- mesh_mcp_server/server.py:163-222 (registration)Dynamically registers tools by fetching agent metadata, filtering by supported agents including 'ElfaTwitterIntelligenceAgent', and creating tool IDs like 'elfatwitterintelligenceagent_search_account' for tools named 'search_account'.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:288-298 (schema)Registers the tool schema (inputSchema from metadata) and description for listing, including for the target tool.@app.list_tools() async def list_tools() -> List[types.Tool]: """List all available tools.""" return [ types.Tool( name=tool_id, description=tool_info["description"], inputSchema=tool_info["parameters"], ) for tool_id, tool_info in self.tool_registry.items() ]
- mesh_mcp_server/server.py:224-262 (helper)Helper function that performs the actual API call to execute the tool on the remote agent.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:39-53 (registration)Configuration list of supported agents that enables 'ElfaTwitterIntelligenceAgent', allowing its tools like 'search_account' to be registered as 'elfatwitterintelligenceagent_search_account'.# Default supported agents DEFAULT_AGENTS = [ "CoinGeckoTokenInfoAgent", "DexScreenerTokenInfoAgent", "ElfaTwitterIntelligenceAgent", "ExaSearchAgent", "TwitterInfoAgent", "AIXBTProjectInfoAgent", "EtherscanAgent", "EvmTokenInfoAgent", "FundingRateAgent", "UnifaiTokenAnalysisAgent", "YahooFinanceAgent", "ZerionWalletAnalysisAgent" ]