elfatwitterintelligenceagent_get_trending_tokens

Instructions

Implementation Reference

• mesh_mcp_server/server.py:301-319 (handler)

  MCP call_tool handler that executes the tool by proxying to the Mesh API after parsing the prefixed tool name to extract agent ID and original tool name.

  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-223 (registration)

  Dynamically registers tools from agent metadata, constructing tool names as '{agent_id.lower()}_{tool_name}', e.g., 'elfatwitterintelligenceagent_get_trending_tokens' for agent 'ElfaTwitterIntelligenceAgent'.

  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:40-53 (registration)

  Configuration listing supported agents, including 'ElfaTwitterIntelligenceAgent' whose tools are registered with the 'elfatwitterintelligenceagent_' prefix.

  DEFAULT_AGENTS = [
      "CoinGeckoTokenInfoAgent",
      "DexScreenerTokenInfoAgent",
      "ElfaTwitterIntelligenceAgent",
      "ExaSearchAgent",
      "TwitterInfoAgent",
      "AIXBTProjectInfoAgent",
      "EtherscanAgent",
      "EvmTokenInfoAgent",
      "FundingRateAgent",
      "UnifaiTokenAnalysisAgent",
      "YahooFinanceAgent",
      "ZerionWalletAnalysisAgent"
  ]

• mesh_mcp_server/server.py:288-298 (registration)

  MCP list_tools decorator that exposes all registered tools, including the target tool with its schema from metadata.

  @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 that sends the tool execution request to the Mesh API endpoint.

  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