Lenses MCP Server

Instructions

Implementation Reference

• src/lenses_mcp/tools/topics.py:415-437 (handler)

  Handler function for the 'update_dataset_topic_description' tool, decorated with @mcp.tool() for registration. It updates the description of a Kafka topic's dataset metadata via an API PUT request.

  @mcp.tool()
  async def update_dataset_topic_description(
      environment: str,
      topic_name: str,
      description: Optional[str] = None
  ) -> Dict[str, Any]:
      """
      Update topic description (in metadata).
      
      Args:
          environment: The environment name.
          topic_name: Name of the topic.
          description: The description of the topic.
      Returns:
          Success message.
      """
      # The description cannot be an empty string so if it is, replace with a null value
      description_payload = { "description": description if description else None }
  
      endpoint = f"/api/v1/environments/{environment}/proxy/api/v1/datasets/kafka/{topic_name}/description"
  
      return await api_client._make_request("PUT", endpoint, description_payload)

• src/lenses_mcp/tools/topics.py:415-415 (registration)

  The @mcp.tool() decorator registers the update_dataset_topic_description tool with the MCP server.

  @mcp.tool()

• src/lenses_mcp/tools/topics.py:421-430 (schema)

  Docstring providing input parameters and return type description, serving as schema documentation.

  """
  Update topic description (in metadata).
  
  Args:
      environment: The environment name.
      topic_name: Name of the topic.
      description: The description of the topic.
  Returns:
      Success message.
  """