MCP Router

Instructions

Search for registered MCP servers based on a query.

Args:
    query: Search query
    top_k: Number of top results to return (default: 3)

Implementation Reference

• mcp_router.py:61-81 (handler)

  The handler function for the 'search_mcp_server' tool, decorated with @mcp.tool() for registration. It calls the discovery_service to search for MCP servers and returns the results as JSON.

  @mcp.tool()
  async def search_mcp_server(query: str, top_k: Optional[int] = 3) -> str:
      """
      Search for registered MCP servers based on a query.
      
      Args:
          query: Search query
          top_k: Number of top results to return (default: 3)
      """
      try:
          # Search for servers
          results = discovery_service.search_server(query, top_k)
          
          # Parse tools JSON strings back to objects
          for result in results:
              result["tools"] = json.loads(result["tools"])
          
          return json.dumps(results, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)

• discovery_service.py:73-108 (helper)

  Supporting helper method in DiscoveryService that performs the actual semantic search using text embeddings from DashScope and vector similarity search in DashVector.

  def search_server(self, query, top_k=3):
      """
      Search for MCP servers based on a query.
      
      Args:
          query (str): Search query
          top_k (int): Number of top results to return (default: 3)
          
      Returns:
          list: List of matching servers with their metadata
      """
      # Generate vector for the query
      query_vector = self._generate_embedding(query)
      
      # Perform vector search
      rsp = self.collection.query(
          query_vector,
          topk=top_k,
          output_fields=['server_description', 'server_endpoint', 'tools']
      )
      
      if not rsp:
          raise Exception("Failed to query DashVector")
      
      # Process results
      results = []
      for doc in rsp.output:
          results.append({
              "server_name": doc.id,
              "server_description": doc.fields['server_description'],
              "server_endpoint": doc.fields['server_endpoint'],
              "tools": doc.fields['tools'],  # Still a JSON string
              "score": doc.score
          })
      
      return results